From 412af679c8d82b6b81efa919bfc524f820243d85 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Wed, 26 Aug 2026 02:10:33 +0200 Subject: [PATCH 1/2] Add response body flow control Expose a thread-safe response body control after final response headers so handlers can suspend, resume, or cancel transport reads without coupling AHC to a streaming API. Integrate the control with HTTP/1.1 and HTTP/2, pause network read timeouts while reads are intentionally suspended, and keep request timeouts active. Make completion, cancellation, callback aborts, and channel teardown restore transport state exactly once. Cover backpressure, cancellation, timeout, callback failures, and connection reuse for both protocols. OpenAI Codex on behalf of Matthias Kurz. Co-Authored-By: OpenAI Codex --- .../org/asynchttpclient/AsyncHandler.java | 16 + .../asynchttpclient/ResponseBodyControl.java | 48 ++ .../netty/handler/AsyncHttpClientHandler.java | 7 +- .../netty/handler/Http2Handler.java | 25 +- .../netty/handler/HttpHandler.java | 22 +- .../handler/NettyResponseBodyControl.java | 162 +++++++ .../netty/timeout/ReadTimeoutTimerTask.java | 7 + .../netty/timeout/TimeoutsHolder.java | 11 +- .../Http2ResponseBodyControlTest.java | 338 ++++++++++++++ .../ResponseBodyControlTest.java | 415 ++++++++++++++++++ .../netty/timeout/TimeoutTimerTaskTest.java | 24 + 11 files changed, 1068 insertions(+), 7 deletions(-) create mode 100644 client/src/main/java/org/asynchttpclient/ResponseBodyControl.java create mode 100644 client/src/main/java/org/asynchttpclient/netty/handler/NettyResponseBodyControl.java create mode 100644 client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java create mode 100644 client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java diff --git a/client/src/main/java/org/asynchttpclient/AsyncHandler.java b/client/src/main/java/org/asynchttpclient/AsyncHandler.java index 22451fe097..9e53d7f3ab 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHandler.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHandler.java @@ -34,6 +34,7 @@ *
    *
  1. {@link #onStatusReceived(HttpResponseStatus)},
    2. *
  2. {@link #onHeadersReceived(HttpHeaders)},
    3. + *
  3. {@link #onResponseBodyStart(ResponseBodyControl)},
    4. *
  4. {@link #onBodyPartReceived(HttpResponseBodyPart)}, which could be invoked multiple times,
    5. *
  5. {@link #onTrailingHeadersReceived(HttpHeaders)}, which is only invoked if trailing HTTP headers are received
    6. *
  6. {@link #onCompleted()}, once the response has been fully read.
    7. @@ -79,6 +80,21 @@ public interface AsyncHandler { */ State onHeadersReceived(HttpHeaders headers) throws Exception; + /** + * Invoked after the final response headers and before any response body parts are delivered. The supplied control + * can suspend and resume transport reads, or cancel the response body. This callback is also invoked for responses + * that have no body. Return {@link State#ABORT} to stop processing from this callback; retain the control and call + * {@link ResponseBodyControl#cancel()} to stop processing asynchronously after this callback returns. + * + * @param control control for this response body. + * @return a {@link State} telling to CONTINUE or ABORT the current processing. + * @throws Exception if something wrong happens + * @since 3.0.14 + */ + default State onResponseBodyStart(ResponseBodyControl control) throws Exception { + return State.CONTINUE; + } + /** * Invoked as soon as some response body part are received. Could be invoked many times. * Beware that, depending on the provider (Netty) this can be notified with empty body parts. diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java new file mode 100644 index 0000000000..7f68c9f02f --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient; + +/** + * Controls transport reads for a response body. + *

    + * The control is thread-safe and remains valid until its response completes. Calls made after completion have no + * effect. + * + * @since 3.0.14 + */ +public interface ResponseBodyControl { + + /** + * Stops requesting additional response bytes from the transport. Body parts that were already read may still be + * delivered to the {@link AsyncHandler}. + *

    + * While reads are suspended, the read timeout is paused but the request timeout remains active. If the request + * timeout is disabled, failing to resume or cancel the response can retain its transport resources indefinitely. + */ + void suspend(); + + /** + * Resumes requesting response bytes after a call to {@link #suspend()}. + */ + void resume(); + + /** + * Stops processing the response body. As with {@link AsyncHandler.State#ABORT}, the handler is completed normally. + * Returning {@code ABORT} is the preferred way to stop from within an {@link AsyncHandler} callback; this method is + * intended for cancellation after the callback has returned, including from another thread. + */ + void cancel(); +} diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java b/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java index d53e98d58e..08b15452e9 100755 --- a/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java @@ -91,6 +91,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { } Channel channel = ctx.channel(); + NettyResponseBodyControl.discardForChannelClose(channel); channelManager.removeAll(channel); Object attribute = Channels.getAttribute(channel); @@ -122,6 +123,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) { } Channel channel = ctx.channel(); + NettyResponseBodyControl.discardForChannelClose(channel); NettyResponseFuture future = null; logger.debug("Unexpected I/O exception on channel {}", channel, cause); @@ -179,7 +181,9 @@ public void channelActive(ChannelHandlerContext ctx) { @Override public void channelReadComplete(ChannelHandlerContext ctx) { - readIfNeeded(ctx); + if (!NettyResponseBodyControl.isSuspended(ctx.channel())) { + readIfNeeded(ctx); + } } /** @@ -196,6 +200,7 @@ private static void readIfNeeded(ChannelHandlerContext ctx) { } void finishUpdate(NettyResponseFuture future, Channel channel, boolean close) { + NettyResponseBodyControl.complete(channel); future.cancelTimeouts(); if (close) { diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java b/client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java index 7c581acbd8..e3626bc6c4 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java @@ -180,14 +180,26 @@ private void handleHttp2HeadersFrame(Http2HeadersFrame headersFrame, Channel cha if (!abort) { abort = handler.onHeadersReceived(responseHeaders) == State.ABORT; } + if (!abort) { + NettyResponseBodyControl control = NettyResponseBodyControl.create( + channel, future::touch, () -> finishUpdate(future, channel, false)); + abort = handler.onResponseBodyStart(control) == State.ABORT; + if (abort) { + NettyResponseBodyControl.complete(channel); + } + } if (abort) { - finishUpdate(future, channel, false); + // cancel() may have completed the future inline from onResponseBodyStart. + if (!future.isDone()) { + finishUpdate(future, channel, false); + } return; } } // If headers frame also ends the stream (no body), finish the response - if (headersFrame.isEndStream()) { + // unless cancel() already completed it inline from onResponseBodyStart. + if (headersFrame.isEndStream() && !future.isDone()) { finishUpdate(future, channel, false); } } @@ -205,6 +217,10 @@ private void handleHttp2DataFrame(Http2DataFrame dataFrame, Channel channel, if (data.isReadable() || last) { HttpResponseBodyPart bodyPart = config.getResponseBodyPartFactory().newResponseBodyPart(data, last); boolean abort = handler.onBodyPartReceived(bodyPart) == State.ABORT; + // cancel() may have completed the future inline from the handler callback. + if (future.isDone()) { + return; + } if (abort || last) { finishUpdate(future, channel, false); } @@ -224,6 +240,10 @@ private void handleHttp2TrailingHeadersFrame(Http2HeadersFrame headersFrame, Cha boolean abort = false; if (!trailingHeaders.isEmpty()) { abort = handler.onTrailingHeadersReceived(trailingHeaders) == State.ABORT; + // cancel() may have completed the future inline from the handler callback. + if (future.isDone()) { + return; + } } if (abort || headersFrame.isEndStream()) { @@ -285,6 +305,7 @@ private void handleHttp2ResetFrame(Http2ResetFrame resetFrame, Channel channel, */ @Override void finishUpdate(NettyResponseFuture future, Channel streamChannel, boolean close) { + NettyResponseBodyControl.complete(streamChannel); future.cancelTimeouts(); // Stream channels are single-use in HTTP/2 — close the stream diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java b/client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java index c09db7b812..b057138039 100755 --- a/client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java @@ -56,6 +56,17 @@ private static boolean abortAfterHandlingHeaders(AsyncHandler handler, HttpHe return !responseHeaders.isEmpty() && handler.onHeadersReceived(responseHeaders) == State.ABORT; } + private boolean abortAfterStartingResponseBody(Channel channel, NettyResponseFuture future, + AsyncHandler handler) throws Exception { + NettyResponseBodyControl control = NettyResponseBodyControl.create( + channel, future::touch, () -> finishUpdate(future, channel, true)); + boolean abort = handler.onResponseBodyStart(control) == State.ABORT; + if (abort) { + NettyResponseBodyControl.complete(channel); + } + return abort; + } + private void handleHttpResponse(final HttpResponse response, final Channel channel, final NettyResponseFuture future, AsyncHandler handler) throws Exception { HttpRequest httpRequest = future.getNettyRequest().getHttpRequest(); if (logger.isDebugEnabled()) { @@ -68,8 +79,11 @@ private void handleHttpResponse(final HttpResponse response, final Channel chann HttpHeaders responseHeaders = response.headers(); if (!interceptors.exitAfterIntercept(channel, future, handler, response, status, responseHeaders)) { - boolean abort = abortAfterHandlingStatus(handler, httpRequest.method(), status) || abortAfterHandlingHeaders(handler, responseHeaders); - if (abort) { + boolean abort = abortAfterHandlingStatus(handler, httpRequest.method(), status) + || abortAfterHandlingHeaders(handler, responseHeaders) + || abortAfterStartingResponseBody(channel, future, handler); + // cancel() may have completed the future inline from onResponseBodyStart. + if (abort && !future.isDone()) { finishUpdate(future, channel, true); } } @@ -92,6 +106,10 @@ private void handleChunk(HttpContent chunk, final Channel channel, final NettyRe if (!abort && (buf.isReadable() || last)) { HttpResponseBodyPart bodyPart = config.getResponseBodyPartFactory().newResponseBodyPart(buf, last); abort = handler.onBodyPartReceived(bodyPart) == State.ABORT; + // cancel() may have completed the future inline from the handler callback. + if (future.isDone()) { + return; + } } if (abort || last) { diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/NettyResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/netty/handler/NettyResponseBodyControl.java new file mode 100644 index 0000000000..3072759cdd --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/netty/handler/NettyResponseBodyControl.java @@ -0,0 +1,162 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.handler; + +import io.netty.channel.Channel; +import io.netty.util.AttributeKey; +import org.asynchttpclient.ResponseBodyControl; +import org.jetbrains.annotations.ApiStatus; + +import java.util.Objects; + +/** + * Netty implementation of {@link ResponseBodyControl}. + */ +@ApiStatus.Internal +public final class NettyResponseBodyControl implements ResponseBodyControl { + + private static final AttributeKey ATTRIBUTE = + AttributeKey.valueOf(NettyResponseBodyControl.class, "control"); + + private final Channel channel; + private final Runnable resumeAction; + private final Runnable cancelAction; + private final boolean previousAutoRead; + + private volatile boolean suspended; + private boolean active = true; + + static NettyResponseBodyControl create(Channel channel, Runnable resumeAction, Runnable cancelAction) { + if (!channel.eventLoop().inEventLoop()) { + throw new IllegalStateException("A response body control must be initialized on its channel event loop"); + } + if (get(channel) != null) { + throw new IllegalStateException("The channel already has a response body control"); + } + + NettyResponseBodyControl control = new NettyResponseBodyControl(channel, resumeAction, cancelAction); + channel.attr(ATTRIBUTE).set(control); + return control; + } + + static NettyResponseBodyControl get(Channel channel) { + return channel != null ? channel.attr(ATTRIBUTE).get() : null; + } + + static void complete(Channel channel) { + NettyResponseBodyControl control = get(channel); + if (control != null) { + control.execute(control::complete0); + } + } + + static void discardForChannelClose(Channel channel) { + NettyResponseBodyControl control = get(channel); + if (control != null) { + control.execute(control::discard0); + } + } + + /** + * Returns whether response reads on {@code channel} are suspended by a response body control. + */ + public static boolean isSuspended(Channel channel) { + NettyResponseBodyControl control = get(channel); + return control != null && control.suspended; + } + + private NettyResponseBodyControl(Channel channel, Runnable resumeAction, Runnable cancelAction) { + this.channel = Objects.requireNonNull(channel, "channel"); + this.resumeAction = Objects.requireNonNull(resumeAction, "resumeAction"); + this.cancelAction = Objects.requireNonNull(cancelAction, "cancelAction"); + previousAutoRead = channel.config().isAutoRead(); + } + + @Override + public void suspend() { + execute(this::suspend0); + } + + @Override + public void resume() { + execute(this::resume0); + } + + @Override + public void cancel() { + execute(this::cancel0); + } + + private void suspend0() { + if (active && !suspended) { + suspended = true; + channel.config().setAutoRead(false); + } + } + + private void resume0() { + if (!active || !suspended) { + return; + } + + suspended = false; + resumeAction.run(); + if (previousAutoRead) { + channel.config().setAutoRead(true); + } else { + channel.read(); + } + } + + private void cancel0() { + if (!active) { + return; + } + + detach(false); + cancelAction.run(); + } + + private void complete0() { + if (active) { + detach(true); + } + } + + private void discard0() { + if (active) { + // The caller is already tearing down the channel, so restoring its read mode has no purpose. + detach(false); + } + } + + private void detach(boolean restoreAutoRead) { + active = false; + suspended = false; + channel.attr(ATTRIBUTE).compareAndSet(this, null); + if (restoreAutoRead && previousAutoRead && !channel.config().isAutoRead()) { + channel.config().setAutoRead(true); + } + } + + private void execute(Runnable task) { + if (channel.eventLoop().inEventLoop()) { + task.run(); + } else { + channel.eventLoop().execute(task); + } + } +} diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java index 8b0d4373a1..92ed876f83 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java @@ -17,6 +17,7 @@ import io.netty.util.Timeout; import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.handler.NettyResponseBodyControl; import org.asynchttpclient.netty.request.NettyRequestSender; import org.asynchttpclient.util.StringBuilderPool; @@ -42,6 +43,12 @@ public void run(Timeout timeout) { return; } + if (NettyResponseBodyControl.isSuspended(nettyResponseFuture.channel())) { + done.set(false); + timeoutsHolder.startReadTimeout(this); + return; + } + long now = unpreciseMillisTime(); long currentReadTimeoutInstant = readTimeout + nettyResponseFuture.getLastTouch(); diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index 93f6b26a26..e50fe8fce4 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -81,7 +81,14 @@ public void startReadTimeout() { } } - void startReadTimeout(ReadTimeoutTimerTask task) { + synchronized void startReadTimeout(ReadTimeoutTimerTask task) { + if (cancelled.get()) { + if (task != null) { + task.clean(); + } + return; + } + if (requestTimeout == null || !requestTimeout.isExpired() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) { // only schedule a new readTimeout if the requestTimeout doesn't happen first if (task == null) { @@ -96,7 +103,7 @@ void startReadTimeout(ReadTimeoutTimerTask task) { } } - public void cancel() { + public synchronized void cancel() { if (cancelled.compareAndSet(false, true)) { if (requestTimeout != null) { requestTimeout.cancel(); diff --git a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java new file mode 100644 index 0000000000..cbf230b129 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -0,0 +1,338 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient; + +import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.http2.DefaultHttp2DataFrame; +import io.netty.handler.codec.http2.DefaultHttp2Headers; +import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame; +import io.netty.handler.codec.http2.Http2FrameCodecBuilder; +import io.netty.handler.codec.http2.Http2HeadersFrame; +import io.netty.handler.codec.http2.Http2MultiplexHandler; +import io.netty.handler.codec.http2.Http2StreamChannel; +import io.netty.handler.ssl.ApplicationProtocolConfig; +import io.netty.handler.ssl.ApplicationProtocolNames; +import io.netty.handler.ssl.SslContext; +import io.netty.handler.ssl.SslContextBuilder; +import io.netty.pkitesting.CertificateBuilder; +import io.netty.pkitesting.X509Bundle; +import io.netty.util.AttributeKey; +import io.netty.util.CharsetUtil; +import io.netty.util.ReferenceCountUtil; +import io.netty.util.concurrent.GlobalEventExecutor; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@ExtendWith(NettyLeakDetectorExtension.class) +public class Http2ResponseBodyControlTest { + + private static final AttributeKey CONNECTION_ID = + AttributeKey.valueOf("response-body-control-h2-connection-id"); + private static final int FRAME_SIZE = 16 * 1024; + private static final int FRAME_COUNT = 16; + + private final AtomicInteger connectionCount = new AtomicInteger(); + private final CountDownLatch largeResponseQueued = new CountDownLatch(1); + private final CompletableFuture largeResponseWritten = new CompletableFuture<>(); + private final CountDownLatch cancelledStreamClosed = new CountDownLatch(1); + + private NioEventLoopGroup serverGroup; + private Channel serverChannel; + private ChannelGroup serverChildChannels; + private SslContext serverSslContext; + private int serverPort; + + @BeforeEach + public void prepareServer() throws Exception { + X509Bundle bundle = new CertificateBuilder() + .subject("CN=localhost") + .setIsCertificateAuthority(true) + .buildSelfSigned(); + serverSslContext = SslContextBuilder.forServer(bundle.toKeyManagerFactory()) + .applicationProtocolConfig(new ApplicationProtocolConfig( + ApplicationProtocolConfig.Protocol.ALPN, + ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE, + ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT, + ApplicationProtocolNames.HTTP_2)) + .build(); + + serverGroup = new NioEventLoopGroup(1); + serverChildChannels = new DefaultChannelGroup("response-body-control-http2", GlobalEventExecutor.INSTANCE); + serverChannel = new ServerBootstrap() + .group(serverGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(Channel channel) { + serverChildChannels.add(channel); + channel.attr(CONNECTION_ID).set(connectionCount.incrementAndGet()); + channel.pipeline() + .addLast(serverSslContext.newHandler(channel.alloc())) + .addLast(Http2FrameCodecBuilder.forServer().build()) + .addLast(new Http2MultiplexHandler(new ChannelInitializer() { + @Override + protected void initChannel(Http2StreamChannel streamChannel) { + serverChildChannels.add(streamChannel); + streamChannel.pipeline().addLast(new StreamingServerHandler()); + } + })); + } + }) + .bind(0) + .sync() + .channel(); + serverPort = ((InetSocketAddress) serverChannel.localAddress()).getPort(); + } + + @AfterEach + public void stopServer() throws InterruptedException { + if (serverChildChannels != null) { + serverChildChannels.close().sync(); + } + if (serverChannel != null) { + serverChannel.close().sync(); + } + if (serverGroup != null) { + serverGroup.shutdownGracefully(0, 100, MILLISECONDS).sync(); + } + ReferenceCountUtil.release(serverSslContext); + } + + @Test + public void suspensionAppliesHttp2FlowControlAndParentIsReused() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture request = client.prepareGet(url("/large")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + assertTrue(largeResponseQueued.await(5, SECONDS)); + assertThrows(TimeoutException.class, () -> largeResponseWritten.get(250, MILLISECONDS), + "suspension must eventually exhaust the HTTP/2 receive window"); + assertTrue(handler.bodyBytes.get() < (long) FRAME_SIZE * FRAME_COUNT, + "the full response must not be delivered while suspended"); + + control.resume(); + largeResponseWritten.get(5, SECONDS); + assertSame(handler, request.get(5, SECONDS)); + assertEquals((long) FRAME_SIZE * FRAME_COUNT, handler.bodyBytes.get()); + assertEquals(2, handler.protocolMajorVersion.get()); + + Response pooled = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", pooled.getResponseBody()); + assertEquals(1, connectionCount.get(), "the next stream must reuse the HTTP/2 parent connection"); + assertNull(handler.throwable.get()); + } + } + + @Test + public void cancellationResetsOnlyTheHttp2Stream() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture request = client.prepareGet(url("/cancel")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + control.resume(); + assertEquals("first", handler.items.poll(5, SECONDS)); + control.cancel(); + + assertSame(handler, request.get(5, SECONDS)); + assertTrue(cancelledStreamClosed.await(5, SECONDS), "cancellation must close the HTTP/2 child stream"); + assertNull(handler.throwable.get()); + + Response sibling = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", sibling.getResponseBody()); + assertEquals(1, connectionCount.get(), "cancellation must preserve the shared HTTP/2 connection"); + } + } + + @Test + public void cancellationFromTerminalBodyCallbackCompletesOnce() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler handler = new RecordingHandler(true); + ListenableFuture request = client.prepareGet(url("/pool")).execute(handler); + + handler.control.get(5, SECONDS).resume(); + + assertSame(handler, request.get(5, SECONDS)); + assertEquals(1, handler.bodyBytes.get()); + assertEquals(1, handler.completionCount.get()); + assertNull(handler.throwable.get()); + + Response sibling = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", sibling.getResponseBody()); + assertEquals(1, connectionCount.get()); + } + } + + private AsyncHttpClient http2Client() { + return asyncHttpClient(config() + .setUseInsecureTrustManager(true) + .setHttp2Enabled(true) + .setHttp2InitialWindowSize(32 * 1024) + .setMaxConnectionsPerHost(1) + .setReadTimeout(Duration.ofMillis(100)) + .setRequestTimeout(Duration.ofSeconds(10))); + } + + private String url(String path) { + return "https://localhost:" + serverPort + path; + } + + private final class StreamingServerHandler extends SimpleChannelInboundHandler { + @Override + protected void channelRead0(ChannelHandlerContext ctx, Object message) { + if (!(message instanceof Http2HeadersFrame)) { + return; + } + Http2HeadersFrame request = (Http2HeadersFrame) message; + String path = request.headers().path().toString(); + switch (path) { + case "/large": + writeHeaders(ctx); + ChannelFuture finalWrite = null; + for (int i = 0; i < FRAME_COUNT; i++) { + ByteBuf content = ctx.alloc().buffer(FRAME_SIZE).writeZero(FRAME_SIZE); + boolean last = i == FRAME_COUNT - 1; + finalWrite = last + ? ctx.writeAndFlush(new DefaultHttp2DataFrame(content, true)) + : ctx.write(new DefaultHttp2DataFrame(content, false)); + } + largeResponseQueued.countDown(); + finalWrite.addListener(result -> { + if (result.isSuccess()) { + largeResponseWritten.complete(null); + } else { + largeResponseWritten.completeExceptionally(result.cause()); + } + }); + break; + case "/cancel": + ctx.channel().closeFuture().addListener(ignored -> cancelledStreamClosed.countDown()); + writeHeaders(ctx); + ctx.writeAndFlush(new DefaultHttp2DataFrame( + Unpooled.copiedBuffer("first", CharsetUtil.US_ASCII), false)); + break; + default: + writeHeaders(ctx); + Integer connectionId = ctx.channel().parent().attr(CONNECTION_ID).get(); + ctx.writeAndFlush(new DefaultHttp2DataFrame( + Unpooled.copiedBuffer(Integer.toString(connectionId), CharsetUtil.US_ASCII), true)); + break; + } + } + + private void writeHeaders(ChannelHandlerContext ctx) { + ctx.write(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers().status("200"), false)); + } + } + + private static final class RecordingHandler implements AsyncHandler { + private final CompletableFuture control = new CompletableFuture<>(); + private final LinkedBlockingQueue items = new LinkedBlockingQueue<>(); + private final AtomicLong bodyBytes = new AtomicLong(); + private final AtomicInteger protocolMajorVersion = new AtomicInteger(); + private final AtomicReference throwable = new AtomicReference<>(); + private final AtomicInteger completionCount = new AtomicInteger(); + private final boolean cancelOnBodyPart; + private ResponseBodyControl responseBodyControl; + + private RecordingHandler() { + this(false); + } + + private RecordingHandler(boolean cancelOnBodyPart) { + this.cancelOnBodyPart = cancelOnBodyPart; + } + + @Override + public State onStatusReceived(HttpResponseStatus responseStatus) { + protocolMajorVersion.set(responseStatus.getProtocolMajorVersion()); + return State.CONTINUE; + } + + @Override + public State onHeadersReceived(io.netty.handler.codec.http.HttpHeaders headers) { + return State.CONTINUE; + } + + @Override + public State onResponseBodyStart(ResponseBodyControl newControl) { + responseBodyControl = newControl; + newControl.suspend(); + control.complete(newControl); + return State.CONTINUE; + } + + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) { + byte[] bytes = bodyPart.getBodyPartBytes(); + bodyBytes.addAndGet(bytes.length); + if (bytes.length > 0) { + items.add(new String(bytes, CharsetUtil.US_ASCII)); + } + if (cancelOnBodyPart) { + responseBodyControl.cancel(); + } + return State.CONTINUE; + } + + @Override + public void onThrowable(Throwable error) { + throwable.compareAndSet(null, error); + } + + @Override + public RecordingHandler onCompleted() { + completionCount.incrementAndGet(); + return this; + } + } +} diff --git a/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java new file mode 100644 index 0000000000..989c315954 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -0,0 +1,415 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient; + +import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.DefaultHttpContent; +import io.netty.handler.codec.http.DefaultHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpResponse; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.codec.http.HttpUtil; +import io.netty.handler.codec.http.LastHttpContent; +import io.netty.util.AttributeKey; +import io.netty.util.CharsetUtil; +import io.netty.util.concurrent.GlobalEventExecutor; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@ExtendWith(NettyLeakDetectorExtension.class) +public class ResponseBodyControlTest { + + private static final AttributeKey CONNECTION_ID = + AttributeKey.valueOf("response-body-control-connection-id"); + + private final AtomicInteger connectionCount = new AtomicInteger(); + private final CompletableFuture responseContext = new CompletableFuture<>(); + private final CountDownLatch cancelledConnectionClosed = new CountDownLatch(1); + + private NioEventLoopGroup serverGroup; + private Channel serverChannel; + private ChannelGroup serverChildChannels; + private int serverPort; + + @BeforeEach + public void startServer() throws InterruptedException { + serverGroup = new NioEventLoopGroup(1); + serverChildChannels = new DefaultChannelGroup("response-body-control-http1", GlobalEventExecutor.INSTANCE); + + serverChannel = new ServerBootstrap() + .group(serverGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(Channel channel) { + serverChildChannels.add(channel); + channel.attr(CONNECTION_ID).set(connectionCount.incrementAndGet()); + channel.pipeline() + .addLast(new HttpServerCodec()) + .addLast(new HttpObjectAggregator(1024)) + .addLast(new StreamingServerHandler()); + } + }) + .bind(0) + .sync() + .channel(); + serverPort = ((InetSocketAddress) serverChannel.localAddress()).getPort(); + } + + @AfterEach + public void stopServer() throws InterruptedException { + if (serverChildChannels != null) { + serverChildChannels.close().sync(); + } + if (serverChannel != null) { + serverChannel.close().sync(); + } + if (serverGroup != null) { + serverGroup.shutdownGracefully(0, 100, MILLISECONDS).sync(); + } + } + + @Test + public void suspensionControlsReadsAndCompletedConnectionIsPooled() throws Exception { + AtomicReference clientChannel = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config() + .setMaxConnectionsPerHost(1) + .setRequestTimeout(Duration.ofSeconds(10)) + .setHttpAdditionalChannelInitializer(clientChannel::set))) { + RecordingHandler handler = new RecordingHandler(true); + ListenableFuture request = client.prepareGet(url("/controlled")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + ChannelHandlerContext server = responseContext.get(5, SECONDS); + + assertFalse(clientChannel.get().config().isAutoRead(), "suspension must pause Netty auto-read"); + writeChunk(server, "one"); + assertNull(handler.items.poll(250, MILLISECONDS), "a suspended response must not read new body bytes"); + + control.resume(); + assertEquals("one", handler.items.poll(5, SECONDS)); + awaitEventLoop(clientChannel.get()); + assertFalse(clientChannel.get().config().isAutoRead(), "the handler suspended the response again"); + + writeChunk(server, "two"); + assertNull(handler.items.poll(250, MILLISECONDS), "the second suspension must also stop reads"); + control.resume(); + assertEquals("two", handler.items.poll(5, SECONDS)); + + writeLast(server); + control.resume(); + assertSame(handler, request.get(5, SECONDS)); + assertTrue(clientChannel.get().config().isAutoRead(), "pooling must restore the channel's read mode"); + + Response pooled = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", pooled.getResponseBody()); + assertEquals(1, connectionCount.get(), "a fully consumed HTTP/1.1 connection must be reused"); + assertNull(handler.throwable.get()); + } + } + + @Test + public void suspensionPausesReadTimeoutAndCancellationClosesConnection() throws Exception { + try (AsyncHttpClient client = asyncHttpClient(config() + .setMaxConnectionsPerHost(1) + .setReadTimeout(Duration.ofMillis(100)) + .setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/cancel")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + assertThrows(TimeoutException.class, () -> request.get(250, MILLISECONDS), + "intentional suspension must pause the network read timeout"); + control.cancel(); + + assertSame(handler, request.get(5, SECONDS), "body cancellation completes the handler normally"); + assertTrue(cancelledConnectionClosed.await(5, SECONDS), "an unread HTTP/1.1 body cannot be pooled"); + assertNull(handler.throwable.get()); + + Response replacement = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("2", replacement.getResponseBody()); + assertEquals(2, connectionCount.get(), "the request after cancellation must use a new connection"); + } + } + + @Test + public void readTimeoutRestartsWhenResponseResumes() throws Exception { + try (AsyncHttpClient client = asyncHttpClient(config() + .setReadTimeout(Duration.ofMillis(100)) + .setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/cancel")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + assertThrows(TimeoutException.class, () -> request.get(250, MILLISECONDS)); + control.resume(); + + ExecutionException failure = assertThrows(ExecutionException.class, () -> request.get(5, SECONDS)); + assertInstanceOf(TimeoutException.class, failure.getCause()); + assertTrue(cancelledConnectionClosed.await(5, SECONDS)); + assertSame(failure.getCause(), handler.throwable.get()); + } + } + + @Test + public void requestTimeoutRemainsActiveWhileSuspended() throws Exception { + try (AsyncHttpClient client = asyncHttpClient(config() + .setReadTimeout(Duration.ofMillis(50)) + .setRequestTimeout(Duration.ofMillis(250)))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/cancel")).execute(handler); + handler.control.get(5, SECONDS); + + ExecutionException failure = assertThrows(ExecutionException.class, () -> request.get(5, SECONDS)); + assertInstanceOf(TimeoutException.class, failure.getCause()); + assertTrue(failure.getCause().getMessage().startsWith("Request timeout")); + assertTrue(cancelledConnectionClosed.await(5, SECONDS)); + assertSame(failure.getCause(), handler.throwable.get()); + } + } + + @Test + public void abortFromResponseBodyStartCompletesNormally() throws Exception { + try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false) { + @Override + public State onResponseBodyStart(ResponseBodyControl newControl) { + return State.ABORT; + } + }; + + assertSame(handler, client.prepareGet(url("/cancel")).execute(handler).get(5, SECONDS)); + assertTrue(handler.items.isEmpty()); + assertEquals(1, handler.completionCount.get()); + assertNull(handler.throwable.get()); + assertTrue(cancelledConnectionClosed.await(5, SECONDS)); + } + } + + @Test + public void exceptionFromResponseBodyStartFailsRequest() throws Exception { + RuntimeException expected = new RuntimeException("response start failed"); + try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false) { + @Override + public State onResponseBodyStart(ResponseBodyControl newControl) { + throw expected; + } + }; + + ListenableFuture request = client.prepareGet(url("/cancel")).execute(handler); + ExecutionException failure = assertThrows(ExecutionException.class, () -> request.get(5, SECONDS)); + assertSame(expected, failure.getCause()); + assertSame(expected, handler.throwable.get()); + assertEquals(0, handler.completionCount.get()); + assertTrue(cancelledConnectionClosed.await(5, SECONDS)); + } + } + + @Test + public void cancellationFromTerminalBodyCallbackCompletesOnce() throws Exception { + AtomicReference callbackControl = new AtomicReference<>(); + try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false) { + @Override + public State onResponseBodyStart(ResponseBodyControl newControl) { + callbackControl.set(newControl); + return State.CONTINUE; + } + + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) { + State state = super.onBodyPartReceived(bodyPart); + callbackControl.get().cancel(); + return state; + } + }; + + assertSame(handler, client.prepareGet(url("/pool")).execute(handler).get(5, SECONDS)); + assertEquals("1", handler.items.poll(5, SECONDS)); + assertEquals(1, handler.completionCount.get()); + assertNull(handler.throwable.get()); + + Response replacement = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("2", replacement.getResponseBody()); + assertEquals(2, connectionCount.get()); + } + } + + @Test + public void responseBodyControlIsProvidedForAnEmptyResponse() throws Exception { + try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false); + ListenableFuture request = client.prepareGet(url("/empty")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + control.resume(); + + assertSame(handler, request.get(5, SECONDS)); + assertTrue(handler.items.isEmpty()); + assertNull(handler.throwable.get()); + } + } + + private String url(String path) { + return "http://localhost:" + serverPort + path; + } + + private static void awaitEventLoop(Channel channel) throws InterruptedException { + channel.eventLoop().submit(() -> { + }).sync(); + } + + private static void writeChunk(ChannelHandlerContext ctx, String value) throws InterruptedException { + ctx.executor().submit(() -> ctx.writeAndFlush( + new DefaultHttpContent(Unpooled.copiedBuffer(value, CharsetUtil.US_ASCII)))).sync(); + } + + private static void writeLast(ChannelHandlerContext ctx) throws InterruptedException { + ctx.executor().submit(() -> ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)).sync(); + } + + private final class StreamingServerHandler extends SimpleChannelInboundHandler { + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) { + switch (request.uri()) { + case "/controlled": + writeStreamingHeaders(ctx); + responseContext.complete(ctx); + break; + case "/cancel": + ctx.channel().closeFuture().addListener(ignored -> cancelledConnectionClosed.countDown()); + writeStreamingHeaders(ctx); + responseContext.complete(ctx); + break; + case "/empty": + DefaultFullHttpResponse emptyResponse = + new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.EMPTY_BUFFER); + HttpUtil.setContentLength(emptyResponse, 0); + HttpUtil.setKeepAlive(emptyResponse, true); + ctx.writeAndFlush(emptyResponse); + break; + default: + ByteBuf content = Unpooled.copiedBuffer( + Integer.toString(ctx.channel().attr(CONNECTION_ID).get()), CharsetUtil.US_ASCII); + DefaultFullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, content); + HttpUtil.setContentLength(response, content.readableBytes()); + HttpUtil.setKeepAlive(response, true); + ctx.writeAndFlush(response); + break; + } + } + + private void writeStreamingHeaders(ChannelHandlerContext ctx) { + HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + HttpUtil.setTransferEncodingChunked(response, true); + HttpUtil.setKeepAlive(response, true); + ctx.writeAndFlush(response); + } + } + + private static class RecordingHandler implements AsyncHandler { + private final boolean suspendEveryPart; + private final CompletableFuture control = new CompletableFuture<>(); + private final LinkedBlockingQueue items = new LinkedBlockingQueue<>(); + private final AtomicReference throwable = new AtomicReference<>(); + private final AtomicInteger completionCount = new AtomicInteger(); + private ResponseBodyControl responseBodyControl; + + private RecordingHandler(boolean suspendEveryPart) { + this.suspendEveryPart = suspendEveryPart; + } + + @Override + public State onStatusReceived(HttpResponseStatus responseStatus) { + return State.CONTINUE; + } + + @Override + public State onHeadersReceived(io.netty.handler.codec.http.HttpHeaders headers) { + return State.CONTINUE; + } + + @Override + public State onResponseBodyStart(ResponseBodyControl newControl) { + responseBodyControl = newControl; + newControl.suspend(); + control.complete(newControl); + return State.CONTINUE; + } + + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) { + if (suspendEveryPart) { + responseBodyControl.suspend(); + } + if (bodyPart.length() > 0) { + items.add(new String(bodyPart.getBodyPartBytes(), CharsetUtil.US_ASCII)); + } + return State.CONTINUE; + } + + @Override + public void onThrowable(Throwable error) { + throwable.compareAndSet(null, error); + } + + @Override + public RecordingHandler onCompleted() { + completionCount.incrementAndGet(); + return this; + } + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java index 2a5f5e2059..f13ce731f8 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java @@ -25,6 +25,7 @@ import java.net.InetSocketAddress; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; public class TimeoutTimerTaskTest { @@ -84,4 +85,27 @@ public void run(io.netty.util.Timeout timeout) { task.appendRemoteAddress(sb); assertTrue(sb.toString().contains(":8080"), sb.toString()); } + + @Test + public void cancelledHolderCleansReschedulingReadTimeout() { + Request request = new RequestBuilder().setUrl("http://example.com").build(); + NettyResponseFuture future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(org.asynchttpclient.Response response) { + return null; + } + }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + TimeoutsHolder timeoutsHolder = new TimeoutsHolder( + null, future, null, new DefaultAsyncHttpClientConfig.Builder().build(), null); + ReadTimeoutTimerTask task = new ReadTimeoutTimerTask(future, null, timeoutsHolder, 1_000); + + // Model cancel() racing after run() marked the task done but before it tries to reschedule itself. + task.done.set(true); + timeoutsHolder.cancel(); + task.done.set(false); + timeoutsHolder.startReadTimeout(task); + + assertNull(task.nettyResponseFuture); + assertTrue(task.done.get()); + } } From 00ada81dc31754fa8cec2a15f6101c2bc4c63bf6 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Wed, 26 Aug 2026 02:12:03 +0200 Subject: [PATCH 2/2] Keep suspended HTTP/2 streams independent Return connection-level receive credit as DATA frames arrive while preserving per-stream flow control for application backpressure. This prevents one suspended response from starving siblings on the same multiplexed connection. Document that aggregate queued data can scale with the number of suspended streams and point users to the initial-window and concurrent-stream controls. Preserve Netty's client shutdown behavior in the custom frame-codec builder. Cover sibling progress, repeated cancellation, and the retained per-stream window bound. OpenAI Codex on behalf of Matthias Kurz. Co-Authored-By: OpenAI Codex --- README.md | 8 ++ .../AsyncHttpClientConfig.java | 7 ++ .../asynchttpclient/ResponseBodyControl.java | 7 ++ .../netty/channel/ChannelManager.java | 26 ++++- .../Http2ResponseBodyControlTest.java | 104 ++++++++++++++++-- 5 files changed, 140 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index b19bef1621..b6eadd0258 100644 --- a/README.md +++ b/README.md @@ -339,6 +339,14 @@ AsyncHttpClient client = asyncHttpClient(config() .setHttp2CleartextEnabled(true)); // h2c prior knowledge ``` +When a handler suspends a response with `ResponseBodyControl`, the HTTP/2 +per-stream window remains the buffering bound for that response. AHC continues +returning connection-level credit so the suspended stream cannot stall sibling +streams. Consequently, aggregate queued response data can scale with the number +of concurrently suspended streams. Use `http2InitialWindowSize` and +`http2MaxConcurrentStreams` together when an application needs a tighter +aggregate bound. + To force HTTP/1.1, disable HTTP/2: ```java diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 7304626083..a5a289a9c4 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -327,6 +327,10 @@ default boolean isHttp2Enabled() { } /** + * This is also the per-stream flow-control bound for response data queued while a + * {@link ResponseBodyControl} is suspended. Aggregate queued data can scale with the number of concurrent suspended + * streams; use {@link #getHttp2MaxConcurrentStreams()} to bound that concurrency. + * * @return the HTTP/2 initial window size in bytes, defaults to 16777216 (16 MiB) */ default int getHttp2InitialWindowSize() { @@ -355,6 +359,9 @@ default int getHttp2MaxHeaderListSize() { } /** + * This setting can be combined with {@link #getHttp2InitialWindowSize()} to bound response data queued for + * concurrently suspended HTTP/2 streams. + * * @return the HTTP/2 max concurrent streams per connection, -1 means unlimited (server-controlled) */ default int getHttp2MaxConcurrentStreams() { diff --git a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java index 7f68c9f02f..8abec9d45a 100644 --- a/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java +++ b/client/src/main/java/org/asynchttpclient/ResponseBodyControl.java @@ -31,6 +31,13 @@ public interface ResponseBodyControl { *

    * While reads are suspended, the read timeout is paused but the request timeout remains active. If the request * timeout is disabled, failing to resume or cancel the response can retain its transport resources indefinitely. + *

    + * For HTTP/2, AHC continues returning connection-level flow-control credit so a suspended stream cannot block + * sibling streams. The per-stream window still applies, so roughly + * {@link AsyncHttpClientConfig#getHttp2InitialWindowSize()} bytes can be queued for each suspended stream. Aggregate + * buffering can therefore scale with the number of concurrent suspended streams; applications can bound it with + * {@link AsyncHttpClientConfig#getHttp2InitialWindowSize()} and + * {@link AsyncHttpClientConfig#getHttp2MaxConcurrentStreams()}. */ void suspend(); diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index f305fb3f34..39cf60abe7 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -37,6 +37,8 @@ import io.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder; import io.netty.handler.codec.http.websocketx.WebSocketFrameAggregator; import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler; +import io.netty.handler.codec.http2.DefaultHttp2Connection; +import io.netty.handler.codec.http2.DefaultHttp2LocalFlowController; import io.netty.handler.codec.http2.DefaultHttp2ResetFrame; import io.netty.handler.codec.http2.Http2Error; import io.netty.handler.codec.http2.Http2FrameCodec; @@ -1079,7 +1081,7 @@ public void upgradePipelineToHttp2(ChannelPipeline pipeline) { // Netty's default and a pushing server could trip a connection-level PROTOCOL_ERROR. .pushEnabled(false); - Http2FrameCodec frameCodec = Http2FrameCodecBuilder.forClient() + Http2FrameCodec frameCodec = new ClientHttp2FrameCodecBuilder() .initialSettings(settings) .build(); @@ -1284,6 +1286,28 @@ private static final class ConnectionCounts { private long idleConnectionCount; } + private static final class ClientHttp2FrameCodecBuilder extends Http2FrameCodecBuilder { + + private ClientHttp2FrameCodecBuilder() { + // Http2FrameCodecBuilder.forClient() sets this through its package-private constructor. This subclass must + // use the protected no-argument constructor, so set it explicitly to preserve the client factory behavior. + gracefulShutdownTimeoutMillis(0); + + DefaultHttp2Connection connection = new DefaultHttp2Connection(false); + // Refill shared credit on receipt so a suspended stream cannot starve siblings. Per-stream windows retain + // application backpressure, at the deliberate cost that aggregate queued data can scale with the number of + // suspended streams. ResponseBodyControl documents the relevant configuration bounds. + connection.local().flowController(new DefaultHttp2LocalFlowController( + connection, DefaultHttp2LocalFlowController.DEFAULT_WINDOW_UPDATE_RATIO, true)); + connection(connection); + } + + @Override + public boolean isServer() { + return false; + } + } + public boolean isOpen() { return channelPool.isOpen(); } diff --git a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java index cbf230b129..3d8364adbe 100644 --- a/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -66,6 +66,7 @@ import static org.asynchttpclient.Dsl.asyncHttpClient; import static org.asynchttpclient.Dsl.config; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -78,11 +79,14 @@ public class Http2ResponseBodyControlTest { AttributeKey.valueOf("response-body-control-h2-connection-id"); private static final int FRAME_SIZE = 16 * 1024; private static final int FRAME_COUNT = 16; + private static final int SIBLING_FRAME_COUNT = 64; + private static final int CANCELLATION_ATTEMPTS = 8; private final AtomicInteger connectionCount = new AtomicInteger(); private final CountDownLatch largeResponseQueued = new CountDownLatch(1); private final CompletableFuture largeResponseWritten = new CompletableFuture<>(); private final CountDownLatch cancelledStreamClosed = new CountDownLatch(1); + private final LinkedBlockingQueue cancelledLargeStreamClosed = new LinkedBlockingQueue<>(); private NioEventLoopGroup serverGroup; private Channel serverChannel; @@ -193,10 +197,63 @@ public void cancellationResetsOnlyTheHttp2Stream() throws Exception { } } + @Test + public void suspendedStreamDoesNotStallSiblingStream() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler suspendedHandler = new RecordingHandler(); + ListenableFuture suspendedRequest = + client.prepareGet(url("/large")).execute(suspendedHandler); + ResponseBodyControl control = suspendedHandler.control.get(5, SECONDS); + + assertTrue(largeResponseQueued.await(5, SECONDS)); + assertThrows(TimeoutException.class, () -> suspendedRequest.get(250, MILLISECONDS)); + + Response sibling = client.prepareGet(url("/large-sibling")) + .setReadTimeout(Duration.ofSeconds(5)) + .execute() + .get(10, SECONDS); + assertEquals((long) FRAME_SIZE * SIBLING_FRAME_COUNT, sibling.getResponseBodyAsBytes().length); + assertEquals(1, connectionCount.get(), "a suspended stream must not stall a sibling on the same connection"); + assertFalse(largeResponseWritten.isDone(), + "connection-level refills must not consume the suspended stream's flow-control window"); + + control.cancel(); + assertSame(suspendedHandler, suspendedRequest.get(5, SECONDS)); + assertNull(suspendedHandler.throwable.get()); + } + } + + @Test + public void repeatedCancellationReturnsHttp2ConnectionWindow() throws Exception { + try (AsyncHttpClient client = http2Client()) { + for (int i = 0; i < CANCELLATION_ATTEMPTS; i++) { + RecordingHandler handler = new RecordingHandler(true); + ListenableFuture request = + client.prepareGet(url("/cancel-large")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + control.resume(); + handler.firstBodyPart.get(5, SECONDS); + control.cancel(); + + assertSame(handler, request.get(5, SECONDS)); + assertTrue(Boolean.TRUE.equals(cancelledLargeStreamClosed.poll(5, SECONDS))); + assertNull(handler.throwable.get()); + } + + Response sibling = client.prepareGet(url("/large-sibling")) + .setReadTimeout(Duration.ofSeconds(5)) + .execute() + .get(10, SECONDS); + assertEquals((long) FRAME_SIZE * SIBLING_FRAME_COUNT, sibling.getResponseBodyAsBytes().length); + assertEquals(1, connectionCount.get(), "cancelled streams must return connection-level flow-control credit"); + } + } + @Test public void cancellationFromTerminalBodyCallbackCompletesOnce() throws Exception { try (AsyncHttpClient client = http2Client()) { - RecordingHandler handler = new RecordingHandler(true); + RecordingHandler handler = new RecordingHandler(false, true); ListenableFuture request = client.prepareGet(url("/pool")).execute(handler); handler.control.get(5, SECONDS).resume(); @@ -237,14 +294,7 @@ protected void channelRead0(ChannelHandlerContext ctx, Object message) { switch (path) { case "/large": writeHeaders(ctx); - ChannelFuture finalWrite = null; - for (int i = 0; i < FRAME_COUNT; i++) { - ByteBuf content = ctx.alloc().buffer(FRAME_SIZE).writeZero(FRAME_SIZE); - boolean last = i == FRAME_COUNT - 1; - finalWrite = last - ? ctx.writeAndFlush(new DefaultHttp2DataFrame(content, true)) - : ctx.write(new DefaultHttp2DataFrame(content, false)); - } + ChannelFuture finalWrite = writeFrames(ctx, FRAME_COUNT); largeResponseQueued.countDown(); finalWrite.addListener(result -> { if (result.isSuccess()) { @@ -260,6 +310,15 @@ protected void channelRead0(ChannelHandlerContext ctx, Object message) { ctx.writeAndFlush(new DefaultHttp2DataFrame( Unpooled.copiedBuffer("first", CharsetUtil.US_ASCII), false)); break; + case "/cancel-large": + ctx.channel().closeFuture().addListener(ignored -> cancelledLargeStreamClosed.offer(Boolean.TRUE)); + writeHeaders(ctx); + writeFrames(ctx, FRAME_COUNT); + break; + case "/large-sibling": + writeHeaders(ctx); + writeFrames(ctx, SIBLING_FRAME_COUNT); + break; default: writeHeaders(ctx); Integer connectionId = ctx.channel().parent().attr(CONNECTION_ID).get(); @@ -272,6 +331,18 @@ protected void channelRead0(ChannelHandlerContext ctx, Object message) { private void writeHeaders(ChannelHandlerContext ctx) { ctx.write(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers().status("200"), false)); } + + private ChannelFuture writeFrames(ChannelHandlerContext ctx, int frameCount) { + ChannelFuture finalWrite = null; + for (int i = 0; i < frameCount; i++) { + ByteBuf content = ctx.alloc().buffer(FRAME_SIZE).writeZero(FRAME_SIZE); + boolean last = i == frameCount - 1; + finalWrite = last + ? ctx.writeAndFlush(new DefaultHttp2DataFrame(content, true)) + : ctx.write(new DefaultHttp2DataFrame(content, false)); + } + return finalWrite; + } } private static final class RecordingHandler implements AsyncHandler { @@ -280,15 +351,22 @@ private static final class RecordingHandler implements AsyncHandler throwable = new AtomicReference<>(); + private final CompletableFuture firstBodyPart = new CompletableFuture<>(); private final AtomicInteger completionCount = new AtomicInteger(); + private final boolean suspendEveryPart; private final boolean cancelOnBodyPart; private ResponseBodyControl responseBodyControl; private RecordingHandler() { - this(false); + this(false, false); + } + + private RecordingHandler(boolean suspendEveryPart) { + this(suspendEveryPart, false); } - private RecordingHandler(boolean cancelOnBodyPart) { + private RecordingHandler(boolean suspendEveryPart, boolean cancelOnBodyPart) { + this.suspendEveryPart = suspendEveryPart; this.cancelOnBodyPart = cancelOnBodyPart; } @@ -316,7 +394,11 @@ public State onBodyPartReceived(HttpResponseBodyPart bodyPart) { byte[] bytes = bodyPart.getBodyPartBytes(); bodyBytes.addAndGet(bytes.length); if (bytes.length > 0) { + if (suspendEveryPart) { + responseBodyControl.suspend(); + } items.add(new String(bytes, CharsetUtil.US_ASCII)); + firstBodyPart.complete(null); } if (cancelOnBodyPart) { responseBodyControl.cancel();