Break the decode loop if decoder raises an exception to give a chance to close the connection to a user handler

- Fixes: #1161

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

@@ -82,12 +82,12 @@ public abstract class ByteToMessageDecoder
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        ByteBuf in = ctx.inboundByteBuffer();
-        if (in.isReadable()) {
-            callDecode(ctx, in);
-        }
-
         try {
+            ByteBuf in = ctx.inboundByteBuffer();
+            if (in.isReadable()) {
+                callDecode(ctx, in);
+            }
+
             if (ctx.nextInboundMessageBuffer().unfoldAndAdd(decodeLast(ctx, in))) {
                 ctx.fireInboundBufferUpdated();
             }
@@ -97,9 +97,9 @@ public abstract class ByteToMessageDecoder
             } else {
                 ctx.fireExceptionCaught(new DecoderException(t));
             }
+        } finally {
+            ctx.fireChannelInactive();
         }
-
-        ctx.fireChannelInactive();
     }
 
     protected void callDecode(ChannelHandlerContext ctx, ByteBuf in) {
@@ -144,6 +144,8 @@ public abstract class ByteToMessageDecoder
                 } else {
                     ctx.fireExceptionCaught(new DecoderException(t));
                 }
+
+                break;
             }
         }
 

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

@@ -364,13 +364,13 @@ public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        replayable.terminate();
-        ByteBuf in = cumulation;
-        if (in.isReadable()) {
-            callDecode(ctx, in);
-        }
-
         try {
+            replayable.terminate();
+            ByteBuf in = cumulation;
+            if (in.isReadable()) {
+                callDecode(ctx, in);
+            }
+
             if (ctx.nextInboundMessageBuffer().unfoldAndAdd(decodeLast(ctx, replayable))) {
                 ctx.fireInboundBufferUpdated();
             }
@@ -383,9 +383,9 @@ public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
             } else {
                 ctx.fireExceptionCaught(new DecoderException(t));
             }
+        } finally {
+            ctx.fireChannelInactive();
         }
-
-        ctx.fireChannelInactive();
     }
 
     @Override
@@ -458,6 +458,8 @@ public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
                 } else {
                     ctx.fireExceptionCaught(new DecoderException(t));
                 }
+
+                break;
             }
         }