Always decode in an event loop

Fixes #3366
reactor · Jul 25, 2024 · e54b5df · e54b5df
1 parent bbf0131
commit e54b5df
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 6 deletions.
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerOperations.java b/reactor-netty-http/src/main/java/reactor/netty/http/server/HttpServerOperations.java
@@ -602,13 +602,10 @@ public HttpServerResponse compression(boolean compress) {
 		if (!compress) {
 			removeHandler(NettyPipeline.CompressionHandler);
 		}
-		else if (channel().pipeline()
-		                  .get(NettyPipeline.CompressionHandler) == null) {
+		else if (channel().pipeline().get(NettyPipeline.CompressionHandler) == null) {
 			SimpleCompressionHandler handler = new SimpleCompressionHandler();
+			handler.request = nettyRequest;
 			try {
-				//Do not invoke handler.channelRead as it will trigger ctx.fireChannelRead
-				handler.decode(channel().pipeline().context(NettyPipeline.ReactiveBridge), nettyRequest);
-
 				addHandlerFirst(NettyPipeline.CompressionHandler, handler);
 			}
 			catch (Throwable e) {

diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/server/SimpleCompressionHandler.java b/reactor-netty-http/src/main/java/reactor/netty/http/server/SimpleCompressionHandler.java
@@ -25,6 +25,8 @@
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpContentCompressor;
 import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.util.ReferenceCountUtil;
 
 import java.util.ArrayList;
@@ -37,10 +39,19 @@
  */
 final class SimpleCompressionHandler extends HttpContentCompressor {
 
+	boolean decoded;
+	HttpRequest request;
+
 	SimpleCompressionHandler() {
 		super((CompressionOptions[]) null);
 	}
 
+	@Override
+	protected void decode(ChannelHandlerContext ctx, HttpRequest msg, List<Object> out) throws Exception {
+		decoded = true;
+		super.decode(ctx, msg, out);
+	}
+
 	@Override
 	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 			throws Exception {
@@ -49,6 +60,13 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 			super.write(ctx, new DefaultHttpContent((ByteBuf) msg), promise);
 		}
 		else {
+			if (!decoded && msg instanceof HttpResponse) {
+				decode(ctx, request);
+			}
+			if (decoded && request != null && msg instanceof LastHttpContent) {
+				decoded = false;
+				request = null;
+			}
 			super.write(ctx, msg, promise);
 		}
 	}
@@ -63,7 +81,7 @@ void decode(ChannelHandlerContext ctx, HttpRequest msg) {
 				// decode(...) will observe a freed content
 				request = new DefaultHttpRequest(msg.protocolVersion(), msg.method(), msg.uri(), msg.headers());
 			}
-			super.decode(ctx, request, out);
+			decode(ctx, request, out);
 		}
 		catch (DecoderException e) {
 			throw e;