From 41c1ab2e826e3cb0f653a4365dcd29c4b59dc5bf Mon Sep 17 00:00:00 2001
From: ursa <ursaj@users.noreply.github.com>
Date: Mon, 24 Jun 2019 08:07:57 +0100
Subject: [PATCH] Bugfix #9257: WebSocketProtocolHandler does NOT support
 autoRead=false (#9258)

Motivation:

I need to control WebSockets inbound flow manually, when autoRead=false

Modification:

Add missed ctx.read() call into WebSocketProtocolHandler, where read request has been swallowed.

Result:

Fixes #9257
---
 .../websocketx/WebSocketProtocolHandler.java  |  8 +++
 .../WebSocketProtocolHandlerTest.java         | 64 ++++++++++++++++++-
 2 files changed, 70 insertions(+), 2 deletions(-)

diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandler.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandler.java
index 53532cafe9..84f96ea344 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandler.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandler.java
@@ -48,15 +48,23 @@ abstract class WebSocketProtocolHandler extends MessageToMessageDecoder<WebSocke
         if (frame instanceof PingWebSocketFrame) {
             frame.content().retain();
             ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content()));
+            readIfNeeded(ctx);
             return;
         }
         if (frame instanceof PongWebSocketFrame && dropPongFrames) {
+            readIfNeeded(ctx);
             return;
         }
 
         out.add(frame.retain());
     }
 
+    private static void readIfNeeded(ChannelHandlerContext ctx) {
+        if (!ctx.channel().config().isAutoRead()) {
+            ctx.read();
+        }
+    }
+
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         ctx.fireExceptionCaught(cause);
diff --git a/codec-http/src/test/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandlerTest.java b/codec-http/src/test/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandlerTest.java
index af7498222c..b59cc1e38f 100644
--- a/codec-http/src/test/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandlerTest.java
+++ b/codec-http/src/test/java/io/netty/handler/codec/http/websocketx/WebSocketProtocolHandlerTest.java
@@ -19,9 +19,10 @@ package io.netty.handler.codec.http.websocketx;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
-import io.netty.util.CharsetUtil;
+import io.netty.handler.flow.FlowControlHandler;
 import org.junit.Test;
 
+import static io.netty.util.CharsetUtil.UTF_8;
 import static org.junit.Assert.*;
 
 /**
@@ -31,7 +32,7 @@ public class WebSocketProtocolHandlerTest {
 
     @Test
     public void testPingFrame() {
-        ByteBuf pingData = Unpooled.copiedBuffer("Hello, world", CharsetUtil.UTF_8);
+        ByteBuf pingData = Unpooled.copiedBuffer("Hello, world", UTF_8);
         EmbeddedChannel channel = new EmbeddedChannel(new WebSocketProtocolHandler() { });
 
         PingWebSocketFrame inputMessage = new PingWebSocketFrame(pingData);
@@ -45,6 +46,65 @@ public class WebSocketProtocolHandlerTest {
         assertFalse(channel.finish());
     }
 
+    @Test
+    public void testPingPongFlowControlWhenAutoReadIsDisabled() {
+        String text1 = "Hello, world #1";
+        String text2 = "Hello, world #2";
+        String text3 = "Hello, world #3";
+        String text4 = "Hello, world #4";
+
+        EmbeddedChannel channel = new EmbeddedChannel();
+        channel.config().setAutoRead(false);
+        channel.pipeline().addLast(new FlowControlHandler());
+        channel.pipeline().addLast(new WebSocketProtocolHandler() { });
+
+        // When
+        assertFalse(channel.writeInbound(
+            new PingWebSocketFrame(Unpooled.copiedBuffer(text1, UTF_8)),
+            new TextWebSocketFrame(text2),
+            new TextWebSocketFrame(text3),
+            new PingWebSocketFrame(Unpooled.copiedBuffer(text4, UTF_8))
+        ));
+
+        // Then - no messages were handled or propagated
+        assertNull(channel.readInbound());
+        assertNull(channel.readOutbound());
+
+        // When
+        channel.read();
+
+        // Then - pong frame was written to the outbound
+        PongWebSocketFrame response1 = channel.readOutbound();
+        assertEquals(text1, response1.content().toString(UTF_8));
+
+        // And - one requested message was handled and propagated inbound
+        TextWebSocketFrame message2 = channel.readInbound();
+        assertEquals(text2, message2.text());
+
+        // And - no more messages were handled or propagated
+        assertNull(channel.readInbound());
+        assertNull(channel.readOutbound());
+
+        // When
+        channel.read();
+
+        // Then - one requested message was handled and propagated inbound
+        TextWebSocketFrame message3 = channel.readInbound();
+        assertEquals(text3, message3.text());
+
+        // And - no more messages were handled or propagated
+        // Precisely, ping frame 'text4' was NOT read or handled.
+        // It would be handle ONLY on the next 'channel.read()' call.
+        assertNull(channel.readInbound());
+        assertNull(channel.readOutbound());
+
+        // Cleanup
+        response1.release();
+        message2.release();
+        message3.release();
+        assertFalse(channel.finish());
+    }
+
     @Test
     public void testPongFrameDropFrameFalse() {
         EmbeddedChannel channel = new EmbeddedChannel(new WebSocketProtocolHandler(false) { });