Repository: netty/netty (Java)

Issue:
WebSocket server loss one frame, maybe caused by ByteToMessageDecoder trigger a unexpected read() when readCompleted fired
We're creating a relay-like application with WebSocket. When the WebSocket client sends an upgrade request to the server, the server sets the channel's autoRead to false (or config child channel autoRead to false, and manually trigger first read()). When the upstream connection is ready (which is started by another type of client), and the relay handler would be ok, set autoRead to true.

### Expected behavior
No `channelRead` fired after setting autoRead to false

### Actual behavior
A frame was read and then discard in TailContext

### Steps to reproduce

See the following junit test:
1. start testWebsocketServer
2. start testWebsocketServerClient
3. watch the server's log
```
18:46:25.176 [nioEventLoopGroup-2-1] DEBUG i.n.channel.DefaultChannelPipeline - Discarded inbound message TextWebSocketFrame(data: PooledUnsafeDirectByteBuf(ridx: 0, widx: 7, cap: 7)) that reached at the tail of the pipeline. Please check your pipeline configuration.
18:46:25.176 [nioEventLoopGroup-2-1] DEBUG i.n.channel.DefaultChannelPipeline - Discarded message pipeline : [wsencoder, wsdecoder, io.netty.handler.codec.http.websocketx.Utf8FrameValidator, WebSocketServerProtocolHandler#0, DefaultChannelPipeline$TailContext#0]. Channel : [id: 0xc1bcc332, L:/127.0.0.1:8100 - R:/127.0.0.1:64710].
18:46:27.149 [nioEventLoopGroup-2-1] INFO  com.example.alexlx.WebSocketTest - read: hello 1
18:46:27.179 [nioEventLoopGroup-2-1] INFO  com.example.alexlx.WebSocketTest - read: hello 2
```

### Minimal yet complete reproducer code (or URL to code)

```java
package com.example.alexlx;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolConfig;
import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;


public class WebSocketTest {
    private static final Logger log = LoggerFactory.getLogger(WebSocketTest.class);
    @Test
    public void testWebsocketServer() {
        NioEventLoopGroup group = new NioEventLoopGroup(1);
        new ServerBootstrap()
                .group(group)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.AUTO_READ, false)
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(
                                new HttpServerCodec(),
                                new HttpObjectAggregator(1024 * 1024),
                                new WebSocketServerProtocolHandler("/ws"),
                                new SimpleUserEventChannelHandler<WebSocketServerProtocolHandler.HandshakeComplete>() {
                                    @Override
                                    protected void eventReceived(ChannelHandlerContext ctx, WebSocketServerProtocolHandler.HandshakeComplete evt) throws Exception {
                                        log.info("handshake {}", evt);
                                        ctx.pipeline().remove(this);
                                        // Simulate the relay connection established in 3 seconds
                                        ctx.executor().schedule(
                                                new Runnable() {
                                                    @Override
                                                    public void run() {
                                                        ctx.channel().config().setAutoRead(true);
                                                        ctx.pipeline().addLast(new SimpleChannelInboundHandler<TextWebSocketFrame>() {
                                                            @Override
                                                            protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
                                                                log.info("read: " + msg.text());
                                                            }
                                                        });
                                                    }
                                                },
                                                3, TimeUnit.SECONDS
                                        );
                                    }
                                }
                        );
                        ch.read();

                    }
                })
                .bind(8100).syncUninterruptibly().channel().closeFuture().syncUninterruptibly();
    }

    @Test
    public void testWebsocketServerClient() {
        new Bootstrap()
                .group(new NioEventLoopGroup(1))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(
                                new HttpClientCodec(),
                                new HttpObjectAggregator(1024 * 1024),
                                new WebSocketClientProtocolHandler(
                                        WebSocketClientProtocolConfig.newBuilder()
                                                .webSocketUri("ws://127.0.0.1:8100/ws")
                                                .build()),
                                new SimpleUserEventChannelHandler<WebSocketClientProtocolHandler.ClientHandshakeStateEvent>() {
                                    @Override
                                    protected void eventReceived(ChannelHandlerContext ctx, WebSocketClientProtocolHandler.ClientHandshakeStateEvent evt) throws Exception {
                                        if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) {
                                            log.info("HANDSHAKE_COMPLETE");
                                            ctx.channel().eventLoop()
                                                    .scheduleAtFixedRate(new Runnable() {
                                                        int i = 0;
                                                        @Override
                                                        public void run() {
                                                            log.info("write {}, writable={}", i, ch.isWritable());
                                                            ch.writeAndFlush(new TextWebSocketFrame("hello " + (i++)));
                                                            if (i >= 10) {
                                                                ctx.close();
                                                            }
                                                        }
                                                    }, 1, 1, TimeUnit.SECONDS);
                                        }
                                    }
                                }
                        );
                    }
                })
                .connect("127.0.0.1", 8100)
                .syncUninterruptibly()
                .channel().closeFuture().syncUninterruptibly();
    }
}
```

### Netty version
4.1.52
### JVM version (e.g. `java -version`)
1.8
### OS version (e.g. `uname -a`)
windows 10 or mac os x 10.14


Hints:

Patch:
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
--- a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
@@ -162,6 +162,8 @@ public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in)
      */
     private boolean firedChannelRead;
 
+    private boolean selfFiredChannelRead;
+
     /**
      * A bitmask where the bits are defined as
      * <ul>
@@ -268,6 +270,7 @@ protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception { }
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (msg instanceof ByteBuf) {
+            selfFiredChannelRead = true;
             CodecOutputList out = CodecOutputList.newInstance();
             try {
                 first = cumulation == null;
@@ -329,7 +332,7 @@ static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int
     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
         numReads = 0;
         discardSomeReadBytes();
-        if (!firedChannelRead && !ctx.channel().config().isAutoRead()) {
+        if (selfFiredChannelRead && !firedChannelRead && !ctx.channel().config().isAutoRead()) {
             ctx.read();
         }
         firedChannelRead = false;


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
--- a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
@@ -25,6 +25,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.channel.socket.ChannelInputShutdownEvent;
 import io.netty.util.internal.PlatformDependent;
@@ -38,6 +39,7 @@
 import static io.netty.buffer.Unpooled.wrappedBuffer;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -434,17 +436,18 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, Byte
         }
     }
 
-    @Test
-    public void testDoesNotOverRead() {
-        class ReadInterceptingHandler extends ChannelOutboundHandlerAdapter {
-            private int readsTriggered;
+    private static final class ReadInterceptingHandler extends ChannelOutboundHandlerAdapter {
+        private int readsTriggered;
 
-            @Override
-            public void read(ChannelHandlerContext ctx) throws Exception {
-                readsTriggered++;
-                super.read(ctx);
-            }
+        @Override
+        public void read(ChannelHandlerContext ctx) throws Exception {
+            readsTriggered++;
+            super.read(ctx);
         }
+    }
+
+    @Test
+    public void testDoesNotOverRead() {
         ReadInterceptingHandler interceptor = new ReadInterceptingHandler();
 
         EmbeddedChannel channel = new EmbeddedChannel();
@@ -538,4 +541,25 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
         assertBuffer(Unpooled.wrappedBuffer(bytes), (ByteBuf) channel.readInbound());
         assertNull(channel.readInbound());
     }
+
+    @Test
+    void testUnexpectRead() {
+        EmbeddedChannel channel = new EmbeddedChannel();
+        channel.config().setAutoRead(false);
+        ReadInterceptingHandler interceptor = new ReadInterceptingHandler();
+        channel.pipeline().addLast(
+                interceptor,
+                new SimpleChannelInboundHandler<ByteBuf>() {
+                    @Override
+                    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
+                        ctx.pipeline().replace(this, "fix", new FixedLengthFrameDecoder(3));
+                    }
+                }
+        );
+
+        assertFalse(channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{1})));
+        assertEquals(0, interceptor.readsTriggered);
+        assertNotNull(channel.pipeline().get(FixedLengthFrameDecoder.class));
+        assertFalse(channel.finish());
+    }
 }