Throw a PrematureChannelClosureException if the channel was closed before all responses were received for the sent requests. See #256

src/main/java/org/jboss/netty/handler/codec/PrematureChannelClosureException.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jboss.netty.handler.codec;
+
+/**
+ * Exception which should get thrown if a Channel got closed before it is expected 
+ */
+public class PrematureChannelClosureException extends Exception {
+    
+    /**
+     * 
+     */
+    private static final long serialVersionUID = 233460005724966593L;
+
+    public PrematureChannelClosureException() {
+        super();
+    }
+    
+    public PrematureChannelClosureException(String msg) {
+        super(msg);
+    }
+    
+    
+    public PrematureChannelClosureException(String msg, Throwable t) {
+        super(msg, t);
+    }
+    
+    
+    public PrematureChannelClosureException(Throwable t) {
+        super(t);
+    }
+    
+
+
+}

src/main/java/org/jboss/netty/handler/codec/http/HttpClientCodec.java

@@ -16,13 +16,16 @@
 package org.jboss.netty.handler.codec.http;
 
 import java.util.Queue;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelDownstreamHandler;
 import org.jboss.netty.channel.ChannelEvent;
 import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
 import org.jboss.netty.channel.ChannelUpstreamHandler;
+import org.jboss.netty.handler.codec.PrematureChannelClosureException;
 import org.jboss.netty.util.internal.QueueFactory;
 
 /**
@@ -33,6 +36,10 @@ import org.jboss.netty.util.internal.QueueFactory;
  * {@link HttpResponseDecoder} to learn what additional state management needs
  * to be done for <tt>HEAD</tt> and <tt>CONNECT</tt> and why
  * {@link HttpResponseDecoder} can not handle it by itself.
+ * 
+ * If the {@link Channel} gets closed and there are requests missing for a response
+ * a {@link PrematureChannelClosureException} is thrown.
+ * 
  * @see HttpServerCodec
  *
  * @apiviz.has org.jboss.netty.handler.codec.http.HttpResponseDecoder
@@ -49,6 +56,7 @@ public class HttpClientCodec implements ChannelUpstreamHandler,
 
     private final HttpRequestEncoder encoder = new Encoder();
     private final HttpResponseDecoder decoder;
+    private final AtomicLong requestResponseCounter = new AtomicLong(0);
 
     /**
      * Creates a new instance with the default decoder options
@@ -89,7 +97,18 @@ public class HttpClientCodec implements ChannelUpstreamHandler,
             if (msg instanceof HttpRequest && !done) {
                 queue.offer(((HttpRequest) msg).getMethod());
             }
-            return super.encode(ctx, channel, msg);
+            
+            Object obj =  super.encode(ctx, channel, msg);
+            
+            // check if the request is chunked if so do not increment
+            if (msg instanceof HttpRequest && !((HttpRequest) msg).isChunked()) {
+                requestResponseCounter.incrementAndGet();
+            } else if (msg instanceof HttpChunk && ((HttpChunk) msg).isLast()) {
+                // increment as its the last chunk
+                requestResponseCounter.incrementAndGet();
+            }
+            
+            return obj;
         }
     }
 
@@ -105,7 +124,18 @@ public class HttpClientCodec implements ChannelUpstreamHandler,
             if (done) {
                 return buffer.readBytes(actualReadableBytes());
             } else {
-                return super.decode(ctx, channel, buffer, state);
+                Object msg = super.decode(ctx, channel, buffer, state);
+                
+                if (msg != null) {
+                    // check if its a HttpMessage and its not chunked
+                    if (msg instanceof HttpMessage && !((HttpMessage) msg).isChunked()) {
+                        requestResponseCounter.decrementAndGet();
+                    } else if (msg instanceof HttpChunk && ((HttpChunk) msg).isLast()) {
+                        requestResponseCounter.decrementAndGet();
+                    }
+                }
+               
+                return msg;
             }
         }
 
@@ -160,5 +190,15 @@ public class HttpClientCodec implements ChannelUpstreamHandler,
 
             return super.isContentAlwaysEmpty(msg);
         }
+        
+        @Override
+        public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+            super.channelClosed(ctx, e);
+            
+            long missingResponses = requestResponseCounter.get();
+            if (missingResponses > 0) {
+                throw new PrematureChannelClosureException("Channel closed but still missing " + missingResponses + " response(s)");
+            }
+        }
     }
 }

src/main/java/org/jboss/netty/handler/codec/package-info.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Base package for codecs
+ *
+ */
+package org.jboss.netty.handler.codec;