Motivation: (#10326)

The current FLowControlHandler keeps a flag to track whether a read() call is pending.
This could lead to a scenario where you call read multiple times when the queue is empty,
and when the FlowControlHandler Queue starts getting messages, channelRead will be fired only once,
when we should've fired x many times, once for each time the handlers downstream called read().

Modifications:

Minor change to replace the boolean flag with a counter and adding a unit test for this scenario.

Result:

I used TDD, so I wrote the test, made sure it's failing, then updated the code and re-ran the test
to make sure it's successful after the changes.

Co-authored-by: Kareem Ali <kali@localhost.localdomain>

handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java

@@ -74,12 +74,17 @@ public class FlowControlHandler extends ChannelDuplexHandler {
 
     private ChannelConfig config;
 
-    private boolean shouldConsume;
+    private int readRequestCount;
 
     public FlowControlHandler() {
         this(true);
     }
 
+    /**
+     * @param releaseMessages If {@code false}, the handler won't release the buffered messages
+     *                        when the handler is removed.
+     *
+     */
     public FlowControlHandler(boolean releaseMessages) {
         this.releaseMessages = releaseMessages;
     }
@@ -140,7 +145,7 @@ public class FlowControlHandler extends ChannelDuplexHandler {
             // It seems no messages were consumed. We need to read() some
             // messages from upstream and once one arrives it need to be
             // relayed to downstream to keep the flow going.
-            shouldConsume = true;
+            ++readRequestCount;
             ctx.read();
         }
     }
@@ -156,8 +161,8 @@ public class FlowControlHandler extends ChannelDuplexHandler {
         // We just received one message. Do we need to relay it regardless
         // of the auto reading configuration? The answer is yes if this
         // method was called as a result of a prior read() call.
-        int minConsume = shouldConsume ? 1 : 0;
-        shouldConsume = false;
+        int minConsume = Math.min(readRequestCount, queue.size());
+        readRequestCount -= minConsume;
 
         dequeue(ctx, minConsume);
     }

handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java

@@ -304,7 +304,7 @@ public class FlowControlHandlerTest {
             // channelRead(2)
             peer.config().setAutoRead(true);
             setAutoReadLatch1.countDown();
-            assertTrue(msgRcvLatch1.await(1L, SECONDS));
+            assertTrue(msgRcvLatch2.await(1L, SECONDS));
 
             // channelRead(3)
             peer.config().setAutoRead(true);
@@ -353,7 +353,7 @@ public class FlowControlHandlerTest {
 
             // Write the message
             client.writeAndFlush(newOneMessage())
-                .syncUninterruptibly();
+                  .syncUninterruptibly();
 
             // channelRead(1)
             peer.read();
@@ -373,6 +373,63 @@ public class FlowControlHandlerTest {
         }
     }
 
+    /**
+     * The {@link FlowControlHandler} will keep track of read calls when
+     * when read is called multiple times when the FlowControlHandler queue is empty.
+     */
+    @Test
+    public void testTrackReadCallCount() throws Exception {
+        final Exchanger<Channel> peerRef = new Exchanger<Channel>();
+        final CountDownLatch msgRcvLatch1 = new CountDownLatch(1);
+        final CountDownLatch msgRcvLatch2 = new CountDownLatch(2);
+        final CountDownLatch msgRcvLatch3 = new CountDownLatch(3);
+
+        ChannelInboundHandlerAdapter handler = new ChannelDuplexHandler() {
+            @Override
+            public void channelActive(ChannelHandlerContext ctx) throws Exception {
+                ctx.fireChannelActive();
+                peerRef.exchange(ctx.channel(), 1L, SECONDS);
+            }
+
+            @Override
+            public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                msgRcvLatch1.countDown();
+                msgRcvLatch2.countDown();
+                msgRcvLatch3.countDown();
+            }
+        };
+
+        FlowControlHandler flow = new FlowControlHandler();
+        Channel server = newServer(false, flow, handler);
+        Channel client = newClient(server.localAddress());
+        try {
+            // The client connection on the server side
+            Channel peer = peerRef.exchange(null, 1L, SECONDS);
+
+            // Confirm that the queue is empty
+            assertTrue(flow.isQueueEmpty());
+            // Request read 3 times
+            peer.read();
+            peer.read();
+            peer.read();
+
+            // Write the message
+            client.writeAndFlush(newOneMessage())
+                  .syncUninterruptibly();
+
+            // channelRead(1)
+            assertTrue(msgRcvLatch1.await(1L, SECONDS));
+            // channelRead(2)
+            assertTrue(msgRcvLatch2.await(1L, SECONDS));
+            // channelRead(3)
+            assertTrue(msgRcvLatch3.await(1L, SECONDS));
+            assertTrue(flow.isQueueEmpty());
+        } finally {
+            client.close();
+            server.close();
+        }
+    }
+
     @Test
     public void testReentranceNotCausesNPE() throws Throwable {
         final Exchanger<Channel> peerRef = new Exchanger<Channel>();