Wie implementiere ich eine von Flow kontrollierte Netty-Pipeline sowohl mit einem FlowControlHandler als auch mit HTTPCo ⇐ Java

[Anonymous]

Duplikatieren https://github.com/netty/netty/issues/15053:
Ich versuche, eine durchflusskontrollierte Netty-Pipeline mit (unter anderem) ein FlowControlHandler in der Nähe der Oberfläche zu erstellen. Flow-kontrolliert in dem Sinne, dass es pro Anruf an read () mehrere Nachrichten ausgibt. Ich habe das Problem auf eine unerwartete Wechselwirkung zwischen FlowControlHandler und httpContentDeCompressor isoliert, die im folgenden Test in der Netty-Test-Suite reproduziert werden:

Code: Select all

index d9f5cd5b8d..f8edf63a29 100644
--- a/codec-http/src/test/java/io/netty/handler/codec/http/HttpContentDecompressorTest.java
+++ b/codec-http/src/test/java/io/netty/handler/codec/http/HttpContentDecompressorTest.java
@@ -20,6 +20,7 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.flow.FlowControlHandler;
import org.junit.jupiter.api.Test;

import java.util.concurrent.atomic.AtomicInteger;
@@ -70,4 +71,33 @@ public class HttpContentDecompressorTest {
assertEquals(2, readCalled.get());
assertFalse(channel.finishAndReleaseAll());
}
+
+    @Test
+    public void testFlowController() {
+        final AtomicInteger messagesEmitted = new AtomicInteger();
+        final EmbeddedChannel channel = new EmbeddedChannel(
+                new FlowControlHandler(),
+                new HttpContentDecompressor(),
+                new ChannelInboundHandlerAdapter() {
+                    @Override
+                    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                        messagesEmitted.incrementAndGet();
+                    }
+                });
+
+        channel.config().setAutoRead(false);
+
+        final HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+        response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
+
+        assertFalse(channel.writeInbound(response));
+        assertFalse(channel.writeInbound(new DefaultHttpContent(Unpooled.buffer(1))));
+        assertFalse(channel.writeInbound(new DefaultHttpContent(Unpooled.buffer(1))));
+
+        assertEquals(1, messagesEmitted.get());
+        channel.read();
+        assertEquals(2, messagesEmitted.get());
+        channel.read();
+        assertEquals(3, messagesEmitted.get());
+    }
}

Speziell mit deaktivierter Autoread und einem FlowControlHandler Ich würde erwarten, dass die Pipeline genau eine Nachricht pro Read () ausgibt. Tatsächlich emittiert sie jedoch heute alle drei Nachrichten gleichzeitig:

Code: Select all

org.opentest4j.AssertionFailedError:
Expected :1
Actual   :3


at org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
at org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
at org.junit.jupiter.api.AssertEquals.failNotEqual(AssertEquals.java:197)
at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:150)
at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:145)
at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:527)
at io.netty.handler.codec.http.HttpContentDecompressorTest.testFlowController(HttpContentDecompressorTest.java:97)
...

Dies geschieht, weil der FlowControlHandler seine eigenen Lesen vollständig Nachrichten erstellt, wenn es seine Warteschlange erfolgt, aber auch die lesende vollständige Nachricht aus dem Upstream, wenn seine interne Warteschlange leer ist. Das erste derartige Ereignis legt httpContentDeCoder#Needread auf true fest und dann löst die zweite von ihnen einen anderen ctx.read () . HttpContentDeCoder sollte in der Lage sein, diese Ereignisse zu verarbeiten, ohne einen anderen ctx.read () auszulösen, oder wenn dieses Gesamtverhalten erwartet wird und ich es nur irgendwie falsch halte. Eine dieser Bestandteile für sich allein, aber wenn sie kombiniert werden, nicht.