Merge #3236 into 1.2.0-M2

reactor · May 13, 2024 · 5f44f2d · 5f44f2d
2 parents f977f73 + 982eb5c
commit 5f44f2d
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 8 deletions.
diff --git a/...or-netty-http/src/main/java/reactor/netty/http/server/Http2StreamBridgeServerHandler.java b/...or-netty-http/src/main/java/reactor/netty/http/server/Http2StreamBridgeServerHandler.java
@@ -177,6 +177,16 @@ else if (!pendingResponse) {
 	@SuppressWarnings("FutureReturnValueIgnored")
 	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
 		if (msg instanceof ByteBuf) {
+			if (!pendingResponse) {
+				if (HttpServerOperations.log.isDebugEnabled()) {
+					HttpServerOperations.log.debug(
+							format(ctx.channel(), "Dropped HTTP content, since response has been sent already: {}"), msg);
+				}
+				((ByteBuf) msg).release();
+				promise.setSuccess();
+				return;
+			}
+
 			//"FutureReturnValueIgnored" this is deliberate
 			ctx.write(new DefaultHttpContent((ByteBuf) msg), promise);
 		}

diff --git a/reactor-netty-http/src/test/java/reactor/netty/http/server/HttpServerTests.java b/reactor-netty-http/src/test/java/reactor/netty/http/server/HttpServerTests.java
@@ -111,6 +111,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.reactivestreams.Publisher;
@@ -1048,7 +1049,8 @@ void testDropPublisherConnectionClose() throws Exception {
 				(req, out) -> {
 					req.addHeader("Connection", "close");
 					return out;
-				});
+				},
+				HttpProtocol.HTTP11);
 		assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
 		assertThat(ReferenceCountUtil.refCnt(data)).isEqualTo(0);
 	}
@@ -1063,12 +1065,14 @@ void testDropMessageConnectionClose() throws Exception {
 				(req, out) -> {
 					req.addHeader("Connection", "close");
 					return out;
-				});
+				},
+				HttpProtocol.HTTP11);
 		assertThat(ReferenceCountUtil.refCnt(data)).isEqualTo(0);
 	}
 
-	@Test
-	void testDropPublisher_1() throws Exception {
+	@ParameterizedTest
+	@EnumSource(value = HttpProtocol.class, names = {"HTTP11", "H2C"})
+	void testDropPublisher_1(HttpProtocol protocol) throws Exception {
 		CountDownLatch latch = new CountDownLatch(1);
 		ByteBuf data = ByteBufAllocator.DEFAULT.buffer();
 		data.writeCharSequence("test", Charset.defaultCharset());
@@ -1077,7 +1081,8 @@ void testDropPublisher_1() throws Exception {
 				                 .send(Flux.defer(() -> Flux.just(data, data.retain(), data.retain()))
 				                           .doFinally(s -> latch.countDown()))
 				                 .then(),
-				(req, out) -> out);
+				(req, out) -> out,
+				protocol);
 		assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
 		assertThat(ReferenceCountUtil.refCnt(data)).isEqualTo(0);
 	}
@@ -1090,7 +1095,8 @@ void testDropPublisher_2() throws Exception {
 				(req, res) -> res.header("Content-Length", "0")
 				                 .send(Mono.just(data))
 				                 .then(),
-				(req, out) -> out);
+				(req, out) -> out,
+				HttpProtocol.HTTP11);
 		assertThat(ReferenceCountUtil.refCnt(data)).isEqualTo(0);
 	}
 
@@ -1101,23 +1107,27 @@ void testDropMessage() throws Exception {
 		doTestDropData(
 				(req, res) -> res.header("Content-Length", "0")
 				                 .sendObject(data),
-				(req, out) -> out);
+				(req, out) -> out,
+				HttpProtocol.HTTP11);
 		assertThat(ReferenceCountUtil.refCnt(data)).isEqualTo(0);
 	}
 
 	private void doTestDropData(
 			BiFunction<? super HttpServerRequest, ? super
 					HttpServerResponse, ? extends Publisher<Void>> serverFn,
-			BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> clientFn)
+			BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> clientFn,
+			HttpProtocol protocol)
 			throws Exception {
 		disposableServer =
 				createServer()
+				          .protocol(protocol)
 				          .handle(serverFn)
 				          .bindNow(Duration.ofSeconds(30));
 
 		CountDownLatch latch = new CountDownLatch(1);
 		String response =
 				createClient(disposableServer.port())
+				          .protocol(protocol)
 				          .doOnRequest((req, conn) -> conn.onTerminate()
 				                                          .subscribe(null, null, latch::countDown))
 				          .request(HttpMethod.GET)