Repository: netty/netty
Issue: Allow to switch to single decoding mode
Labels: improvement
feature

In our *Decoder implementation we always try to decode as many messages as possible. This is the best performance wise but is sometimes problematic. Especially when you are about to do a protocol upgrade and so will replace the decoder.

To fix this we should provide a way to switch to only decode one message per inboundBuffferUpdated(..) call. This will also fix #864.


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker00.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker00.java
index 040399d3991..6cb8768e10d 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker00.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker00.java
@@ -173,6 +173,8 @@ public ChannelFuture handshake(Channel channel, final ChannelPromise promise) {
         headers.set(Names.CONTENT_LENGTH, key3.length);
         request.data().writeBytes(key3);
 
+        channel.pipeline().get(HttpResponseDecoder.class).setSingleDecode(true);
+
         ChannelFuture future = channel.write(request);
         future.addListener(new ChannelFutureListener() {
             @Override
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker07.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker07.java
index f2ce512e8d4..118a281ef40 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker07.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker07.java
@@ -152,6 +152,8 @@ public ChannelFuture handshake(Channel channel, final ChannelPromise promise) {
             headers.add(customHeaders);
         }
 
+        channel.pipeline().get(HttpResponseDecoder.class).setSingleDecode(true);
+
         ChannelFuture future = channel.write(request);
         future.addListener(new ChannelFutureListener() {
             @Override
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker08.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker08.java
index 21bf020744d..cb227a25b09 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker08.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker08.java
@@ -152,6 +152,8 @@ public ChannelFuture handshake(Channel channel, final ChannelPromise promise) {
             headers.add(customHeaders);
         }
 
+        channel.pipeline().get(HttpResponseDecoder.class).setSingleDecode(true);
+
         ChannelFuture future = channel.write(request);
         future.addListener(new ChannelFutureListener() {
             @Override
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker13.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker13.java
index 55ac1bf38e3..b5adbe6eb28 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker13.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketClientHandshaker13.java
@@ -152,6 +152,8 @@ public ChannelFuture handshake(Channel channel, final ChannelPromise promise) {
             headers.add(customHeaders);
         }
 
+        channel.pipeline().get(HttpResponseDecoder.class).setSingleDecode(true);
+
         ChannelFuture future = channel.write(request);
         future.addListener(new ChannelFutureListener() {
             @Override
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToByteDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToByteDecoder.java
index 334bcb12e5b..31a3cf7e3ee 100644
--- a/codec/src/main/java/io/netty/handler/codec/ByteToByteDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToByteDecoder.java
@@ -44,6 +44,28 @@
  */
 public abstract class ByteToByteDecoder extends ChannelInboundByteHandlerAdapter {
 
+    private volatile boolean singleDecode;
+
+    /**
+     * If set then only one message is decoded on each {@link #inboundBufferUpdated(ChannelHandlerContext)} call.
+     * This may be useful if you need to do some protocol upgrade and want to make sure nothing is mixed up.
+     *
+     * Default is {@code false} as this has performance impacts.
+     */
+    public void setSingleDecode(boolean singleDecode) {
+        this.singleDecode = singleDecode;
+    }
+
+    /**
+     * If {@code true} then only one message is decoded on each
+     * {@link #inboundBufferUpdated(ChannelHandlerContext)} call.
+     *
+     * Default is {@code false} as this has performance impacts.
+     */
+    public boolean isSingleDecode() {
+        return singleDecode;
+    }
+
     @Override
     public void inboundBufferUpdated(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
         callDecode(ctx, in, ctx.nextInboundByteBuffer());
@@ -91,7 +113,7 @@ private void callDecode(ChannelHandlerContext ctx, ByteBuf in, ByteBuf out) {
                     ctx.fireExceptionCaught(new DecoderException(t));
                 }
             }
-            if (oldInSize == in.readableBytes()) {
+            if (oldInSize == in.readableBytes() || isSingleDecode()) {
                 break;
             }
         }
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
index b32b095f572..d82e8c6b23e 100644
--- a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
@@ -44,6 +44,28 @@ public abstract class ByteToMessageDecoder
 
     private ChannelHandlerContext ctx;
 
+    private volatile boolean singleDecode;
+
+    /**
+     * If set then only one message is decoded on each {@link #inboundBufferUpdated(ChannelHandlerContext)} call.
+     * This may be useful if you need to do some protocol upgrade and want to make sure nothing is mixed up.
+     *
+     * Default is {@code false} as this has performance impacts.
+     */
+    public void setSingleDecode(boolean singleDecode) {
+        this.singleDecode = singleDecode;
+    }
+
+    /**
+     * If {@code true} then only one message is decoded on each
+     * {@link #inboundBufferUpdated(ChannelHandlerContext)} call.
+     *
+     * Default is {@code false} as this has performance impacts.
+     */
+    public boolean isSingleDecode() {
+        return singleDecode;
+    }
+
     @Override
     public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
         this.ctx = ctx;
@@ -109,6 +131,9 @@ protected void callDecode(ChannelHandlerContext ctx) {
 
                 if (ChannelHandlerUtil.unfoldAndAdd(ctx, o, true)) {
                     decoded = true;
+                    if (isSingleDecode()) {
+                        break;
+                    }
                 } else {
                     break;
                 }
diff --git a/codec/src/main/java/io/netty/handler/codec/ReplayingDecoder.java b/codec/src/main/java/io/netty/handler/codec/ReplayingDecoder.java
index 02743758d42..76bc1a04003 100644
--- a/codec/src/main/java/io/netty/handler/codec/ReplayingDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ReplayingDecoder.java
@@ -432,6 +432,9 @@ protected void callDecode(ChannelHandlerContext ctx) {
                 // A successful decode
                 if (ChannelHandlerUtil.unfoldAndAdd(ctx, result, true)) {
                     decoded = true;
+                    if (isSingleDecode()) {
+                        break;
+                    }
                 }
             } catch (Throwable t) {
                 if (decoded) {


Test patch: