Repository: netty/netty (Java)

Issue:
FlowControllerHandler swallows read-complete event when auto-read is disabled
### Expected behavior

IdleStateHandler should trigger read-timeout events when registered after FlowControlHandler.

### Actual behavior

IdleStateHandler does NOT trigger read-timeout event.

### Steps to reproduce

Server pipeline:

    p.addLast("flow-control", new FlowControlHandler)
    p.addLast("keep-alive", new PingInactiveHandler(readTimeout))// extends IdleStateHandler
    p.addLast("handler", new AppHandler) // trigger read

When no upstream messages from the client within time limit, then READER_IDLE event should be triggered. However it does NOT happen.

### Netty version

* Version 4.1.31 - no problem, example code works.
* Version 4.1.39 - has a problem, example code does NOT work.

Seems like related to PR #9664 

### JVM version (e.g. `java -version`)

Oracle JDK 1.8.0_221-b27

### OS version (e.g. `uname -a`)

* Windows  7 Enterprise
* Linux gmafxu42 3.0.101-108.98-default #1 SMP Mon Jul 15 13:58:06 UTC 2019 (262a94d) x86_64 x86_64 x86_64 GNU/Linux

### How to fix

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    ---    // Don't relay completion events from upstream as they
    ---    // make no sense in this context. See dequeue() where
    ---    // a new set of completion events is being produced.
    +++    if (isQueueEmpty()) {
    +++        ctx.fireChannelReadComplete();
    +++    }
    }



Hints:
@ursaj sounds legit. Want to do a PR ?

Patch:
diff --git a/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java b/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java
--- a/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java
+++ b/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java
@@ -154,9 +154,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-        // Don't relay completion events from upstream as they
-        // make no sense in this context. See dequeue() where
-        // a new set of completion events is being produced.
+        if (isQueueEmpty()) {
+            ctx.fireChannelReadComplete();
+        } else {
+            // Don't relay completion events from upstream as they
+            // make no sense in this context. See dequeue() where
+            // a new set of completion events is being produced.
+        }
     }
 
     /**


Test patch:
diff --git a/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java b/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java
--- a/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java
+++ b/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java
@@ -29,10 +29,13 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.embedded.EmbeddedChannel;
 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.ByteToMessageDecoder;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.ReferenceCountUtil;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -40,13 +43,14 @@
 
 import java.net.SocketAddress;
 import java.util.List;
+import java.util.Queue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Exchanger;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static java.util.concurrent.TimeUnit.*;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
 
 public class FlowControlHandlerTest {
     private static EventLoopGroup GROUP;
@@ -77,7 +81,7 @@ private static Channel newServer(final boolean autoRead, final ChannelHandler...
             .childOption(ChannelOption.AUTO_READ, autoRead)
             .childHandler(new ChannelInitializer<Channel>() {
                 @Override
-                protected void initChannel(Channel ch) throws Exception {
+                protected void initChannel(Channel ch) {
                     ChannelPipeline pipeline = ch.pipeline();
                     pipeline.addLast(new OneByteToThreeStringsDecoder());
                     pipeline.addLast(handlers);
@@ -419,13 +423,69 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         }
     }
 
+    @Test
+    public void testSwallowedReadComplete() throws Exception {
+        final long delayMillis = 100;
+        final Queue<IdleStateEvent> userEvents = new LinkedBlockingQueue<IdleStateEvent>();
+        final EmbeddedChannel channel = new EmbeddedChannel(false, false,
+            new FlowControlHandler(),
+            new IdleStateHandler(delayMillis, 0, 0, MILLISECONDS),
+            new ChannelInboundHandlerAdapter() {
+                @Override
+                public void channelActive(ChannelHandlerContext ctx) {
+                    ctx.fireChannelActive();
+                    ctx.read();
+                }
+
+                @Override
+                public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                    ctx.fireChannelRead(msg);
+                    ctx.read();
+                }
+
+                @Override
+                public void channelReadComplete(ChannelHandlerContext ctx) {
+                    ctx.fireChannelReadComplete();
+                    ctx.read();
+                }
+
+                @Override
+                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+                    if (evt instanceof IdleStateEvent) {
+                        userEvents.add((IdleStateEvent) evt);
+                    }
+                    ctx.fireUserEventTriggered(evt);
+                }
+            }
+        );
+
+        channel.config().setAutoRead(false);
+        assertFalse(channel.config().isAutoRead());
+
+        channel.register();
+
+        // Reset read timeout by some message
+        assertTrue(channel.writeInbound(Unpooled.EMPTY_BUFFER));
+        channel.flushInbound();
+        assertEquals(Unpooled.EMPTY_BUFFER, channel.readInbound());
+
+        // Emulate 'no more messages in NIO channel' on the next read attempt.
+        channel.flushInbound();
+        assertNull(channel.readInbound());
+
+        Thread.sleep(delayMillis);
+        channel.runPendingTasks();
+        assertEquals(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT, userEvents.poll());
+        assertFalse(channel.finish());
+    }
+
     /**
      * This is a fictional message decoder. It decodes each {@code byte}
      * into three strings.
      */
     private static final class OneByteToThreeStringsDecoder extends ByteToMessageDecoder {
         @Override
-        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
             for (int i = 0; i < in.readableBytes(); i++) {
                 out.add("1");
                 out.add("2");