Add response body flow control

README.md

@@ -339,6 +339,17 @@ 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. While at least
+one response on a connection is suspended, AHC continues returning
+connection-level credit so it cannot stall sibling streams. Connections with no
+active suspension retain the normal 65,535-byte shared connection-window bound.
+Once the last suspension ends, normal connection accounting resumes, although
+credit already returned and data already queued cannot be revoked. Aggregate
+queued response data during suspension can scale with the number of concurrent
+streams. Use `http2InitialWindowSize` and `http2MaxConcurrentStreams` together
+when an application needs a tighter aggregate bound.
+
 To force HTTP/1.1, disable HTTP/2:
 
 ```java

client/src/main/java/org/asynchttpclient/AsyncHandler.java

@@ -34,6 +34,7 @@
  * <ol>
  * <li>{@link #onStatusReceived(HttpResponseStatus)},</li>
  * <li>{@link #onHeadersReceived(HttpHeaders)},</li>
+ * <li>{@link #onResponseBodyStart(ResponseBodyControl)},</li>
  * <li>{@link #onBodyPartReceived(HttpResponseBodyPart)}, which could be invoked multiple times,</li>
  * <li>{@link #onTrailingHeadersReceived(HttpHeaders)}, which is only invoked if trailing HTTP headers are received</li>
  * <li>{@link #onCompleted()}, once the response has been fully read.</li>
@@ -79,6 +80,23 @@ public interface AsyncHandler<T> {
      */
     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.
+     * If the final headers also end the response, suspending cannot defer completion: the control becomes inactive when
+     * this callback returns and later calls have no effect.
+     *
+     * @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.

client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -383,6 +383,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() {
@@ -411,6 +415,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() {

client/src/main/java/org/asynchttpclient/ResponseBodyControl.java

@@ -0,0 +1,62 @@
+/*
+ *    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.
+ * <p>
+ * The control is thread-safe and remains valid until its response completes. Calls made after completion have no
+ * effect.
+ * <p>
+ * A control is also supplied when the final response headers end the response without a body. In that case,
+ * {@link #suspend()} cannot defer completion: the control becomes inactive when
+ * {@link AsyncHandler#onResponseBodyStart(ResponseBodyControl)} returns, and later calls 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}.
+     * If the final response headers already ended the response, this call has no effect on completion.
+     * <p>
+     * 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.
+     * <p>
+     * For HTTP/2, while any response on a connection is suspended, AHC continues returning connection-level
+     * flow-control credit so a suspended stream cannot block sibling streams. Responses on connections with no active
+     * suspension retain the normal shared connection-window bound. The per-stream window always applies, so roughly
+     * {@link AsyncHttpClientConfig#getHttp2InitialWindowSize()} bytes can be queued for each suspended stream. Aggregate
+     * buffering during suspension can therefore scale with the number of concurrent streams. Once the last suspension
+     * ends, normal connection accounting resumes; credit already returned and data already queued cannot be revoked.
+     * Applications can bound buffering with {@link AsyncHttpClientConfig#getHttp2InitialWindowSize()} and
+     * {@link AsyncHttpClientConfig#getHttp2MaxConcurrentStreams()}.
+     */
+    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();
+}

client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java

@@ -0,0 +1,203 @@
+/*
+ *    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;
+
+import io.netty.channel.Channel;
+import org.asynchttpclient.ResponseBodyControl;
+import org.jetbrains.annotations.ApiStatus;
+
+import java.util.Objects;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * Netty implementation of {@link ResponseBodyControl}.
+ */
+@ApiStatus.Internal
+public final class NettyResponseBodyControl implements ResponseBodyControl {
+
+    private final NettyResponseFuture<?> future;
+    private final Channel channel;
+    private final Runnable suspensionStartedAction;
+    private final Runnable suspensionEndedAction;
+    private final Runnable resumeAction;
+    private final Consumer<Boolean> cancelAction;
+    private final boolean previousAutoRead;
+
+    private final AtomicBoolean active = new AtomicBoolean(true);
+    private volatile boolean suspended;
+    private volatile boolean bodyFullyRead;
+
+    public static NettyResponseBodyControl create(NettyResponseFuture<?> future, Channel channel,
+                                                  Runnable resumeAction, Consumer<Boolean> cancelAction) {
+        return create(future, channel, NettyResponseBodyControl::noop, NettyResponseBodyControl::noop,
+                resumeAction, cancelAction);
+    }
+
+    public static NettyResponseBodyControl create(NettyResponseFuture<?> future, Channel channel,
+                                                  Runnable suspensionStartedAction,
+                                                  Runnable suspensionEndedAction,
+                                                  Runnable resumeAction, Consumer<Boolean> cancelAction) {
+        if (!channel.eventLoop().inEventLoop()) {
+            throw new IllegalStateException("A response body control must be initialized on its channel event loop");
+        }
+
+        NettyResponseBodyControl control = new NettyResponseBodyControl(
+                future, channel, suspensionStartedAction, suspensionEndedAction, resumeAction, cancelAction);
+        NettyResponseBodyControl previous = future.replaceResponseBodyControl(control);
+        if (previous != null) {
+            previous.deactivate(true);
+        }
+        return control;
+    }
+
+    public static void complete(NettyResponseFuture<?> future) {
+        NettyResponseBodyControl control = future.responseBodyControl();
+        if (control != null) {
+            control.deactivate(true);
+        }
+    }
+
+    public static void discardForChannelClose(NettyResponseFuture<?> future, Channel channel) {
+        NettyResponseBodyControl control = future.responseBodyControl();
+        if (control != null && control.channel == channel) {
+            control.deactivate(false);
+        }
+    }
+
+    /**
+     * Returns whether response reads for {@code future} are suspended by its current response body control.
+     */
+    public static boolean isSuspended(NettyResponseFuture<?> future) {
+        NettyResponseBodyControl control = future.responseBodyControl();
+        return control != null && control.active.get() && control.suspended;
+    }
+
+    /**
+     * Returns whether {@code future} is suspended on {@code channel}.
+     */
+    public static boolean isSuspended(NettyResponseFuture<?> future, Channel channel) {
+        NettyResponseBodyControl control = future.responseBodyControl();
+        return control != null && control.channel == channel && control.active.get() && control.suspended;
+    }
+
+    /**
+     * Records that the complete HTTP/1.1 response has reached the client before its terminal callbacks run.
+     */
+    public static void markBodyFullyRead(NettyResponseFuture<?> future) {
+        NettyResponseBodyControl control = future.responseBodyControl();
+        if (control != null) {
+            control.bodyFullyRead = true;
+        }
+    }
+
+    private NettyResponseBodyControl(NettyResponseFuture<?> future, Channel channel,
+                                     Runnable suspensionStartedAction, Runnable suspensionEndedAction,
+                                     Runnable resumeAction, Consumer<Boolean> cancelAction) {
+        this.future = Objects.requireNonNull(future, "future");
+        this.channel = Objects.requireNonNull(channel, "channel");
+        this.suspensionStartedAction = Objects.requireNonNull(suspensionStartedAction, "suspensionStartedAction");
+        this.suspensionEndedAction = Objects.requireNonNull(suspensionEndedAction, "suspensionEndedAction");
+        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.get() && !suspended) {
+            suspensionStartedAction.run();
+            suspended = true;
+            channel.config().setAutoRead(false);
+        }
+    }
+
+    private void resume0() {
+        if (!active.get() || !suspended) {
+            return;
+        }
+
+        endSuspension();
+        resumeAction.run();
+        if (previousAutoRead) {
+            channel.config().setAutoRead(true);
+        } else {
+            channel.read();
+        }
+    }
+
+    private void cancel0() {
+        if (!active.compareAndSet(true, false)) {
+            return;
+        }
+
+        future.clearResponseBodyControl(this);
+        detach0(bodyFullyRead);
+        cancelAction.accept(bodyFullyRead);
+    }
+
+    private void deactivate(boolean restoreAutoRead) {
+        if (!active.compareAndSet(true, false)) {
+            return;
+        }
+        future.clearResponseBodyControl(this);
+        execute(() -> detach0(restoreAutoRead));
+    }
+
+    private void detach0(boolean restoreAutoRead) {
+        endSuspension();
+        if (restoreAutoRead && previousAutoRead && !channel.config().isAutoRead()) {
+            channel.config().setAutoRead(true);
+        }
+    }
+
+    private void endSuspension() {
+        if (suspended) {
+            suspended = false;
+            suspensionEndedAction.run();
+        }
+    }
+
+    private void execute(Runnable task) {
+        if (channel.eventLoop().inEventLoop()) {
+            task.run();
+        } else {
+            try {
+                channel.eventLoop().execute(task);
+            } catch (RejectedExecutionException ignored) {
+                // The channel is shutting down, so the control has no transport left to affect.
+            }
+        }
+    }
+
+    private static void noop() {
+    }
+}

client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java

@@ -31,6 +31,7 @@
 import org.asynchttpclient.proxy.ProxyServer;
 import org.asynchttpclient.scram.ScramContext;
 import org.asynchttpclient.uri.Uri;
+import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -88,6 +89,9 @@ public final class NettyResponseFuture<V> implements ListenableFuture<V> {
     private static final AtomicReferenceFieldUpdater<NettyResponseFuture, TimeoutsHolder> TIMEOUTS_HOLDER_FIELD = AtomicReferenceFieldUpdater
             .newUpdater(NettyResponseFuture.class, TimeoutsHolder.class, "timeoutsHolder");
     @SuppressWarnings("rawtypes")
+    private static final AtomicReferenceFieldUpdater<NettyResponseFuture, NettyResponseBodyControl> RESPONSE_BODY_CONTROL_FIELD =
+            AtomicReferenceFieldUpdater.newUpdater(NettyResponseFuture.class, NettyResponseBodyControl.class, "responseBodyControl");
+    @SuppressWarnings("rawtypes")
     private static final AtomicReferenceFieldUpdater<NettyResponseFuture, Object> PARTITION_KEY_LOCK_FIELD = AtomicReferenceFieldUpdater
             .newUpdater(NettyResponseFuture.class, Object.class, "partitionKeyLock");
 
@@ -116,6 +120,8 @@ public final class NettyResponseFuture<V> implements ListenableFuture<V> {
     private volatile int onThrowableCalled;
     @SuppressWarnings("unused")
     private volatile TimeoutsHolder timeoutsHolder;
+    @SuppressWarnings("unused")
+    private volatile @Nullable NettyResponseBodyControl responseBodyControl;
     // partition key, when != null used to release lock in ChannelManager
     private volatile Object partitionKeyLock;
     // volatile where we need CAS ops
@@ -402,6 +408,18 @@ public void cancelTimeouts() {
         }
     }
 
+    @Nullable NettyResponseBodyControl responseBodyControl() {
+        return responseBodyControl;
+    }
+
+    @Nullable NettyResponseBodyControl replaceResponseBodyControl(NettyResponseBodyControl control) {
+        return RESPONSE_BODY_CONTROL_FIELD.getAndSet(this, control);
+    }
+
+    void clearResponseBodyControl(NettyResponseBodyControl control) {
+        RESPONSE_BODY_CONTROL_FIELD.compareAndSet(this, control, null);
+    }
+
     public Request getTargetRequest() {
         return targetRequest;
     }