ByteToMessageDecoder ChannelInputShutdownEvent support

Motivation:
b714297a44 introduced ChannelInputShutdownEvent support for HttpObjectDecoder. However this should have been added to the super class ByteToMessageDecoder, and ByteToMessageDecoder should not propegate a channelInactive event through the pipeline in this case.

Modifications:
- Move the ChannelInputShutdownEvent handling from HttpObjectDecoder to ByteToMessageDecoder
- ByteToMessageDecoder doesn't call ctx.fireChannelInactive() on ChannelInputShutdownEvent

Result:
Half closed events are treated more generically, and don't get translated into a channelInactive pipeline event.

codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java

@@ -19,7 +19,6 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
-import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.codec.DecoderResult;
 import io.netty.handler.codec.TooLongFrameException;
@@ -431,12 +430,7 @@ public abstract class HttpObjectDecoder extends ByteToMessageDecoder {
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-        if (evt instanceof ChannelInputShutdownEvent) {
+        if (evt instanceof HttpExpectationFailedEvent) {
-            // The decodeLast method is invoked when a channelInactive event is encountered.
-            // This method is responsible for ending requests in some situations and must be called
-            // when the input has been shutdown.
-            super.channelInactive(ctx);
-        } else if (evt instanceof HttpExpectationFailedEvent) {
             switch (currentState) {
             case READ_FIXED_LENGTH_CONTENT:
             case READ_VARIABLE_LENGTH_CONTENT:

codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java

@@ -21,6 +21,7 @@ import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.util.internal.RecyclableArrayList;
 import io.netty.util.internal.StringUtil;
 
@@ -305,6 +306,21 @@ public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        channelInputClosed(ctx, true);
+    }
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof ChannelInputShutdownEvent) {
+            // The decodeLast method is invoked when a channelInactive event is encountered.
+            // This method is responsible for ending requests in some situations and must be called
+            // when the input has been shutdown.
+            channelInputClosed(ctx, false);
+        }
+        super.userEventTriggered(ctx, evt);
+    }
+
+    private void channelInputClosed(ChannelHandlerContext ctx, boolean callChannelInactive) throws Exception {
         RecyclableArrayList out = RecyclableArrayList.newInstance();
         try {
             if (cumulation != null) {
@@ -329,7 +345,9 @@ public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
                     // Something was read, call fireChannelReadComplete()
                     ctx.fireChannelReadComplete();
                 }
-                ctx.fireChannelInactive();
+                if (callChannelInactive) {
+                    ctx.fireChannelInactive();
+                }
             } finally {
                 // recycle in all cases
                 out.recycle();