diff --git a/README.md b/README.md index b19bef162..1bbe2803c 100644 --- a/README.md +++ b/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 diff --git a/client/src/main/java/org/asynchttpclient/AsyncHandler.java b/client/src/main/java/org/asynchttpclient/AsyncHandler.java index 22451fe09..b2eb82f17 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,23 @@ 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. + * 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. diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 3adc7a30b..78a8a086b 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/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() { 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 000000000..7ada0f7fa --- /dev/null +++ b/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. + *

    + * The control is thread-safe and remains valid until its response completes. Calls made after completion have no + * effect. + *

    + * 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. + *

    + * 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, 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(); +} diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseBodyControl.java new file mode 100644 index 000000000..65fae4f8b --- /dev/null +++ b/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 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 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 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 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() { + } +} diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index c3616e5bd..48f16ecc6 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/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 implements ListenableFuture { private static final AtomicReferenceFieldUpdater TIMEOUTS_HOLDER_FIELD = AtomicReferenceFieldUpdater .newUpdater(NettyResponseFuture.class, TimeoutsHolder.class, "timeoutsHolder"); @SuppressWarnings("rawtypes") + private static final AtomicReferenceFieldUpdater RESPONSE_BODY_CONTROL_FIELD = + AtomicReferenceFieldUpdater.newUpdater(NettyResponseFuture.class, NettyResponseBodyControl.class, "responseBodyControl"); + @SuppressWarnings("rawtypes") private static final AtomicReferenceFieldUpdater PARTITION_KEY_LOCK_FIELD = AtomicReferenceFieldUpdater .newUpdater(NettyResponseFuture.class, Object.class, "partitionKeyLock"); @@ -116,6 +120,8 @@ public final class NettyResponseFuture implements ListenableFuture { 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; } 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 f305fb3f3..655f05d3c 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -37,8 +37,10 @@ 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.DefaultHttp2ResetFrame; import io.netty.handler.codec.http2.Http2Error; +import io.netty.handler.codec.http2.Http2Exception; import io.netty.handler.codec.http2.Http2FrameCodec; import io.netty.handler.codec.http2.Http2FrameCodecBuilder; import io.netty.handler.codec.http2.Http2MultiplexHandler; @@ -1079,9 +1081,10 @@ 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() - .initialSettings(settings) - .build(); + ClientHttp2FrameCodecBuilder frameCodecBuilder = new ClientHttp2FrameCodecBuilder(); + Http2FrameCodec frameCodec = frameCodecBuilder.initialSettings(settings).build(); + pipeline.channel().attr(SuspensionAwareHttp2LocalFlowController.CHANNEL_KEY) + .set(frameCodecBuilder.flowController()); // Http2MultiplexHandler creates a child channel per HTTP/2 stream. // Server-push streams are rejected with RST_STREAM(REFUSED_STREAM). @@ -1284,6 +1287,62 @@ private static final class ConnectionCounts { private long idleConnectionCount; } + private static final class ClientHttp2FrameCodecBuilder extends Http2FrameCodecBuilder { + + private final SuspensionAwareHttp2LocalFlowController flowController; + + 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); + flowController = new SuspensionAwareHttp2LocalFlowController(connection); + connection.local().flowController(flowController); + connection(connection); + } + + private SuspensionAwareHttp2LocalFlowController flowController() { + return flowController; + } + + @Override + public boolean isServer() { + return false; + } + } + + /** + * Enables connection-window refill for the lifetime of a suspended HTTP/2 response. + */ + public void suspendHttp2ResponseBody(Channel streamChannel) { + SuspensionAwareHttp2LocalFlowController controller = http2FlowController(streamChannel); + try { + controller.suspendResponse(); + } catch (Http2Exception e) { + PlatformDependent.throwException(e); + } + } + + /** + * Restores normal connection-window accounting after an HTTP/2 response stops being suspended. + */ + public void resumeHttp2ResponseBody(Channel streamChannel) { + http2FlowController(streamChannel).resumeResponse(); + } + + private static SuspensionAwareHttp2LocalFlowController http2FlowController(Channel streamChannel) { + Channel parentChannel = streamChannel instanceof Http2StreamChannel + ? ((Http2StreamChannel) streamChannel).parent() + : streamChannel; + SuspensionAwareHttp2LocalFlowController controller = + parentChannel.attr(SuspensionAwareHttp2LocalFlowController.CHANNEL_KEY).get(); + if (controller == null) { + throw new IllegalStateException("HTTP/2 response body flow controller is not installed"); + } + return controller; + } + public boolean isOpen() { return channelPool.isOpen(); } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/SuspensionAwareHttp2LocalFlowController.java b/client/src/main/java/org/asynchttpclient/netty/channel/SuspensionAwareHttp2LocalFlowController.java new file mode 100644 index 000000000..9f6ee647e --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/netty/channel/SuspensionAwareHttp2LocalFlowController.java @@ -0,0 +1,540 @@ +/* + * 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. + */ +/* + * Portions adapted from Netty 4.2.17.Final's DefaultHttp2LocalFlowController. + * Copyright 2014 The Netty Project, licensed under Apache License 2.0. + */ +package org.asynchttpclient.netty.channel; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http2.Http2Connection; +import io.netty.handler.codec.http2.Http2ConnectionAdapter; +import io.netty.handler.codec.http2.Http2Error; +import io.netty.handler.codec.http2.Http2Exception; +import io.netty.handler.codec.http2.Http2Exception.CompositeStreamException; +import io.netty.handler.codec.http2.Http2Exception.StreamException; +import io.netty.handler.codec.http2.Http2FrameWriter; +import io.netty.handler.codec.http2.Http2LocalFlowController; +import io.netty.handler.codec.http2.Http2Stream; +import io.netty.handler.codec.http2.Http2StreamVisitor; +import io.netty.util.AttributeKey; +import io.netty.util.internal.PlatformDependent; + +import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID; +import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_WINDOW_SIZE; +import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_INITIAL_WINDOW_SIZE; +import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_INITIAL_WINDOW_SIZE; +import static io.netty.handler.codec.http2.Http2Error.FLOW_CONTROL_ERROR; +import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR; +import static io.netty.handler.codec.http2.Http2Exception.connectionError; +import static io.netty.handler.codec.http2.Http2Exception.streamError; +import static io.netty.util.internal.ObjectUtil.checkNotNull; +import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero; +import static java.lang.Math.max; +import static java.lang.Math.min; + +/** + * Netty's default local HTTP/2 flow controller adapted to refill connection credit only while at least one response + * on the connection is suspended. Stream credit remains consumption-driven at all times. + * + *

    The implementation follows {@code DefaultHttp2LocalFlowController} as released in Netty 4.2.17.Final. Netty's + * connection auto-refill state is private and fixed at construction time, so it cannot be enabled only for the + * lifetime of a suspended response by composition or subclassing. Because AHC owns this package-private adaptation, + * Netty upgrades must compare it with the corresponding upstream implementation for correctness and security fixes. + * An upstream API for changing connection auto-refill at runtime would allow this class to be removed.

    + * + *

    This class is not thread safe. All methods are invoked on the HTTP/2 connection event loop.

    + */ +final class SuspensionAwareHttp2LocalFlowController implements Http2LocalFlowController { + + static final AttributeKey CHANNEL_KEY = + AttributeKey.valueOf(SuspensionAwareHttp2LocalFlowController.class, "controller"); + + private static final float DEFAULT_WINDOW_UPDATE_RATIO = 0.5f; + + private final Http2Connection connection; + private final Http2Connection.PropertyKey stateKey; + private Http2FrameWriter frameWriter; + private ChannelHandlerContext ctx; + private float windowUpdateRatio; + private int initialWindowSize = DEFAULT_WINDOW_SIZE; + private int suspendedResponses; + + SuspensionAwareHttp2LocalFlowController(Http2Connection connection) { + this.connection = checkNotNull(connection, "connection"); + windowUpdateRatio(DEFAULT_WINDOW_UPDATE_RATIO); + + stateKey = connection.newKey(); + connection.connectionStream().setProperty( + stateKey, new SuspensionAwareConnectionState(connection.connectionStream(), initialWindowSize)); + + connection.addListener(new Http2ConnectionAdapter() { + @Override + public void onStreamAdded(Http2Stream stream) { + stream.setProperty(stateKey, REDUCED_FLOW_STATE); + } + + @Override + public void onStreamActive(Http2Stream stream) { + stream.setProperty(stateKey, new DefaultState(stream, initialWindowSize)); + } + + @Override + public void onStreamClosed(Http2Stream stream) { + try { + FlowState state = state(stream); + int unconsumedBytes = state.unconsumedBytes(); + if (ctx != null && unconsumedBytes > 0 && consumeAllBytes(state, unconsumedBytes)) { + ctx.flush(); + } + } catch (Http2Exception e) { + PlatformDependent.throwException(e); + } finally { + stream.setProperty(stateKey, REDUCED_FLOW_STATE); + } + } + }); + } + + void suspendResponse() throws Http2Exception { + assert ctx != null && ctx.executor().inEventLoop(); + if (suspendedResponses == 0) { + connectionState().startAutoRefill(); + } + suspendedResponses++; + } + + void resumeResponse() { + assert ctx != null && ctx.executor().inEventLoop(); + if (suspendedResponses <= 0) { + throw new IllegalStateException("No suspended HTTP/2 response to resume"); + } + suspendedResponses--; + } + + boolean hasSuspendedResponse() { + return suspendedResponses > 0; + } + + long autoConsumedConnectionBytes() { + return connectionState().autoConsumedBytes; + } + + @Override + public SuspensionAwareHttp2LocalFlowController frameWriter(Http2FrameWriter frameWriter) { + this.frameWriter = checkNotNull(frameWriter, "frameWriter"); + return this; + } + + @Override + public void channelHandlerContext(ChannelHandlerContext ctx) { + this.ctx = checkNotNull(ctx, "ctx"); + } + + @Override + public void initialWindowSize(int newWindowSize) throws Http2Exception { + assert ctx == null || ctx.executor().inEventLoop(); + int delta = newWindowSize - initialWindowSize; + initialWindowSize = newWindowSize; + + WindowUpdateVisitor visitor = new WindowUpdateVisitor(delta); + connection.forEachActiveStream(visitor); + visitor.throwIfError(); + } + + @Override + public int initialWindowSize() { + return initialWindowSize; + } + + @Override + public int windowSize(Http2Stream stream) { + return state(stream).windowSize(); + } + + @Override + public int initialWindowSize(Http2Stream stream) { + return state(stream).initialWindowSize(); + } + + @Override + public void incrementWindowSize(Http2Stream stream, int delta) throws Http2Exception { + assert ctx != null && ctx.executor().inEventLoop(); + FlowState state = state(stream); + state.incrementInitialStreamWindow(delta); + state.writeWindowUpdateIfNeeded(); + } + + @Override + public boolean consumeBytes(Http2Stream stream, int numBytes) throws Http2Exception { + assert ctx != null && ctx.executor().inEventLoop(); + checkPositiveOrZero(numBytes, "numBytes"); + if (numBytes == 0) { + return false; + } + + if (stream != null && !isClosed(stream)) { + if (stream.id() == CONNECTION_STREAM_ID) { + throw new UnsupportedOperationException("Returning bytes for the connection window is not supported"); + } + return consumeAllBytes(state(stream), numBytes); + } + return false; + } + + private boolean consumeAllBytes(FlowState state, int numBytes) throws Http2Exception { + return connectionState().consumeBytes(numBytes) | state.consumeBytes(numBytes); + } + + @Override + public int unconsumedBytes(Http2Stream stream) { + return state(stream).unconsumedBytes(); + } + + private static void checkValidRatio(float ratio) { + if (Double.compare(ratio, 0.0) <= 0 || Double.compare(ratio, 1.0) >= 0) { + throw new IllegalArgumentException("Invalid ratio: " + ratio); + } + } + + public void windowUpdateRatio(float ratio) { + assert ctx == null || ctx.executor().inEventLoop(); + checkValidRatio(ratio); + windowUpdateRatio = ratio; + } + + public float windowUpdateRatio() { + return windowUpdateRatio; + } + + public void windowUpdateRatio(Http2Stream stream, float ratio) throws Http2Exception { + assert ctx != null && ctx.executor().inEventLoop(); + checkValidRatio(ratio); + FlowState state = state(stream); + state.windowUpdateRatio(ratio); + state.writeWindowUpdateIfNeeded(); + } + + public float windowUpdateRatio(Http2Stream stream) throws Http2Exception { + return state(stream).windowUpdateRatio(); + } + + @Override + public void receiveFlowControlledFrame(Http2Stream stream, ByteBuf data, int padding, + boolean endOfStream) throws Http2Exception { + assert ctx != null && ctx.executor().inEventLoop(); + int dataLength = data.readableBytes() + padding; + + SuspensionAwareConnectionState connectionState = connectionState(); + connectionState.receiveFlowControlledFrame(dataLength); + + if (stream != null && !isClosed(stream)) { + FlowState state = state(stream); + state.endOfStream(endOfStream); + state.receiveFlowControlledFrame(dataLength); + } else if (dataLength > 0) { + connectionState.consumeBytes(dataLength); + } + } + + private SuspensionAwareConnectionState connectionState() { + return connection.connectionStream().getProperty(stateKey); + } + + private FlowState state(Http2Stream stream) { + return stream.getProperty(stateKey); + } + + private static boolean isClosed(Http2Stream stream) { + return stream.state() == Http2Stream.State.CLOSED; + } + + private final class SuspensionAwareConnectionState extends DefaultState { + + private long autoConsumedBytes; + + SuspensionAwareConnectionState(Http2Stream stream, int initialWindowSize) { + super(stream, initialWindowSize); + } + + void startAutoRefill() throws Http2Exception { + // DATA may already have reached a stream child channel before its handler calls suspend(). Return that + // outstanding connection credit too, otherwise the pre-suspension bytes could still starve siblings. + int unconsumedBytes = unconsumedBytes(); + if (unconsumedBytes > 0) { + super.consumeBytes(unconsumedBytes); + autoConsumedBytes += unconsumedBytes; + } + } + + @Override + public void receiveFlowControlledFrame(int dataLength) throws Http2Exception { + super.receiveFlowControlledFrame(dataLength); + if (hasSuspendedResponse() && dataLength > 0) { + super.consumeBytes(dataLength); + autoConsumedBytes += dataLength; + } + } + + @Override + public boolean consumeBytes(int numBytes) throws Http2Exception { + int alreadyConsumed = (int) min(autoConsumedBytes, (long) numBytes); + autoConsumedBytes -= alreadyConsumed; + int remaining = numBytes - alreadyConsumed; + return remaining > 0 && super.consumeBytes(remaining); + } + } + + private class DefaultState implements FlowState { + + private final Http2Stream stream; + private int window; + private int processedWindow; + private int initialStreamWindowSize; + private float streamWindowUpdateRatio; + private int lowerBound; + private boolean endOfStream; + + DefaultState(Http2Stream stream, int initialWindowSize) { + this.stream = stream; + window(initialWindowSize); + streamWindowUpdateRatio = windowUpdateRatio; + } + + @Override + public void window(int initialWindowSize) { + assert ctx == null || ctx.executor().inEventLoop(); + window = processedWindow = initialStreamWindowSize = initialWindowSize; + } + + @Override + public int windowSize() { + return window; + } + + @Override + public int initialWindowSize() { + return initialStreamWindowSize; + } + + @Override + public void endOfStream(boolean endOfStream) { + this.endOfStream = endOfStream; + } + + @Override + public float windowUpdateRatio() { + return streamWindowUpdateRatio; + } + + @Override + public void windowUpdateRatio(float ratio) { + assert ctx == null || ctx.executor().inEventLoop(); + streamWindowUpdateRatio = ratio; + } + + @Override + public void incrementInitialStreamWindow(int delta) { + int newValue = (int) min(MAX_INITIAL_WINDOW_SIZE, + max(MIN_INITIAL_WINDOW_SIZE, initialStreamWindowSize + (long) delta)); + initialStreamWindowSize += newValue - initialStreamWindowSize; + } + + @Override + public void incrementFlowControlWindows(int delta) throws Http2Exception { + if (delta > 0 && window > MAX_INITIAL_WINDOW_SIZE - delta) { + throw streamError(stream.id(), FLOW_CONTROL_ERROR, + "Flow control window overflowed for stream: %d", stream.id()); + } + window += delta; + processedWindow += delta; + lowerBound = min(delta, 0); + } + + @Override + public void receiveFlowControlledFrame(int dataLength) throws Http2Exception { + assert dataLength >= 0; + window -= dataLength; + if (window < lowerBound) { + throw streamError(stream.id(), FLOW_CONTROL_ERROR, + "Flow control window exceeded for stream: %d", stream.id()); + } + } + + private void returnProcessedBytes(int delta) throws Http2Exception { + if (processedWindow - delta < window) { + throw streamError(stream.id(), INTERNAL_ERROR, + "Attempting to return too many bytes for stream %d", stream.id()); + } + processedWindow -= delta; + } + + @Override + public boolean consumeBytes(int numBytes) throws Http2Exception { + returnProcessedBytes(numBytes); + return writeWindowUpdateIfNeeded(); + } + + @Override + public int unconsumedBytes() { + return processedWindow - window; + } + + @Override + public boolean writeWindowUpdateIfNeeded() throws Http2Exception { + if (endOfStream || initialStreamWindowSize <= 0 || isClosed(stream)) { + return false; + } + int threshold = (int) (initialStreamWindowSize * streamWindowUpdateRatio); + if (processedWindow <= threshold) { + writeWindowUpdate(); + return true; + } + return false; + } + + private void writeWindowUpdate() throws Http2Exception { + int deltaWindowSize = initialStreamWindowSize - processedWindow; + try { + incrementFlowControlWindows(deltaWindowSize); + } catch (Throwable t) { + throw connectionError(INTERNAL_ERROR, t, + "Attempting to return too many bytes for stream %d", stream.id()); + } + frameWriter.writeWindowUpdate(ctx, stream.id(), deltaWindowSize, ctx.newPromise()); + } + } + + private static final FlowState REDUCED_FLOW_STATE = new FlowState() { + @Override + public int windowSize() { + return 0; + } + + @Override + public int initialWindowSize() { + return 0; + } + + @Override + public void window(int initialWindowSize) { + throw new UnsupportedOperationException(); + } + + @Override + public void incrementInitialStreamWindow(int delta) { + // Required while the peer has not yet acknowledged the stream as active. + } + + @Override + public boolean writeWindowUpdateIfNeeded() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean consumeBytes(int numBytes) { + return false; + } + + @Override + public int unconsumedBytes() { + return 0; + } + + @Override + public float windowUpdateRatio() { + throw new UnsupportedOperationException(); + } + + @Override + public void windowUpdateRatio(float ratio) { + throw new UnsupportedOperationException(); + } + + @Override + public void receiveFlowControlledFrame(int dataLength) { + throw new UnsupportedOperationException(); + } + + @Override + public void incrementFlowControlWindows(int delta) { + // Required while the peer has not yet acknowledged the stream as active. + } + + @Override + public void endOfStream(boolean endOfStream) { + throw new UnsupportedOperationException(); + } + }; + + private interface FlowState { + int windowSize(); + + int initialWindowSize(); + + void window(int initialWindowSize); + + void incrementInitialStreamWindow(int delta); + + boolean writeWindowUpdateIfNeeded() throws Http2Exception; + + boolean consumeBytes(int numBytes) throws Http2Exception; + + int unconsumedBytes(); + + float windowUpdateRatio(); + + void windowUpdateRatio(float ratio); + + void receiveFlowControlledFrame(int dataLength) throws Http2Exception; + + void incrementFlowControlWindows(int delta) throws Http2Exception; + + void endOfStream(boolean endOfStream); + } + + private final class WindowUpdateVisitor implements Http2StreamVisitor { + + private final int delta; + private CompositeStreamException compositeException; + + WindowUpdateVisitor(int delta) { + this.delta = delta; + } + + @Override + public boolean visit(Http2Stream stream) throws Http2Exception { + try { + FlowState state = state(stream); + state.incrementFlowControlWindows(delta); + state.incrementInitialStreamWindow(delta); + } catch (StreamException e) { + if (compositeException == null) { + compositeException = new CompositeStreamException(e.error(), 4); + } + compositeException.add(e); + } + return true; + } + + void throwIfError() throws CompositeStreamException { + if (compositeException != null) { + throw compositeException; + } + } + } +} 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 d53e98d58..7868486fa 100755 --- a/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/AsyncHttpClientHandler.java @@ -24,6 +24,7 @@ import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.exception.ChannelClosedException; import org.asynchttpclient.netty.DiscardEvent; +import org.asynchttpclient.netty.NettyResponseBodyControl; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.OnLastHttpContentCallback; import org.asynchttpclient.netty.channel.ChannelManager; @@ -86,14 +87,19 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + Channel channel = ctx.channel(); + Object attribute = Channels.getAttribute(channel); + NettyResponseFuture controlFuture = responseFuture(attribute); + if (controlFuture != null) { + NettyResponseBodyControl.discardForChannelClose(controlFuture, channel); + } + if (requestSender.isClosed()) { return; } - Channel channel = ctx.channel(); channelManager.removeAll(channel); - Object attribute = Channels.getAttribute(channel); logger.debug("Channel Closed: {} with attribute {}", channel, attribute); if (attribute instanceof OnLastHttpContentCallback) { OnLastHttpContentCallback callback = (OnLastHttpContentCallback) attribute; @@ -113,6 +119,16 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { } } + private static NettyResponseFuture responseFuture(Object attribute) { + if (attribute instanceof NettyResponseFuture) { + return (NettyResponseFuture) attribute; + } + if (attribute instanceof OnLastHttpContentCallback) { + return ((OnLastHttpContentCallback) attribute).future(); + } + return null; + } + @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) { Throwable cause = getCause(e); @@ -138,6 +154,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) { if (hasIOExceptionFilters) { if (!requestSender.applyIoExceptionFiltersAndReplayRequest(future, ChannelClosedException.INSTANCE, channel)) { // Close the channel so the recovering can occurs. + NettyResponseBodyControl.discardForChannelClose(future, channel); Channels.silentlyCloseChannel(channel); } return; @@ -166,6 +183,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) { } } + if (future != null) { + NettyResponseBodyControl.discardForChannelClose(future, channel); + } channelManager.closeChannel(channel); // FIXME not really sure // ctx.fireChannelRead(e); @@ -179,7 +199,15 @@ public void channelActive(ChannelHandlerContext ctx) { @Override public void channelReadComplete(ChannelHandlerContext ctx) { - readIfNeeded(ctx); + Channel channel = ctx.channel(); + if (channel.config().isAutoRead()) { + return; + } + Object attribute = Channels.getAttribute(channel); + if (!(attribute instanceof NettyResponseFuture) + || !NettyResponseBodyControl.isSuspended((NettyResponseFuture) attribute, channel)) { + ctx.read(); + } } /** @@ -196,6 +224,7 @@ private static void readIfNeeded(ChannelHandlerContext ctx) { } void finishUpdate(NettyResponseFuture future, Channel channel, boolean close) { + NettyResponseBodyControl.complete(future); 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 7c581acbd..114f8150e 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/Http2Handler.java @@ -36,6 +36,7 @@ import org.asynchttpclient.AsyncHandler.State; import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.HttpResponseBodyPart; +import org.asynchttpclient.netty.NettyResponseBodyControl; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.NettyResponseStatus; import org.asynchttpclient.netty.channel.ChannelManager; @@ -180,14 +181,26 @@ private void handleHttp2HeadersFrame(Http2HeadersFrame headersFrame, Channel cha if (!abort) { abort = handler.onHeadersReceived(responseHeaders) == State.ABORT; } + if (!abort) { + NettyResponseBodyControl control = NettyResponseBodyControl.create( + future, channel, + () -> channelManager.suspendHttp2ResponseBody(channel), + () -> channelManager.resumeHttp2ResponseBody(channel), + future::touch, ignored -> finishUpdate(future, channel, false)); + abort = handler.onResponseBodyStart(control) == State.ABORT; + } 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 +218,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 +241,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 +306,7 @@ private void handleHttp2ResetFrame(Http2ResetFrame resetFrame, Channel channel, */ @Override void finishUpdate(NettyResponseFuture future, Channel streamChannel, boolean close) { + NettyResponseBodyControl.complete(future); 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 c09db7b81..7eb310178 100755 --- a/client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java @@ -29,9 +29,12 @@ import org.asynchttpclient.AsyncHandler.State; import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.HttpResponseBodyPart; +import org.asynchttpclient.netty.NettyResponseBodyControl; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.NettyResponseStatus; +import org.asynchttpclient.netty.OnLastHttpContentCallback; import org.asynchttpclient.netty.channel.ChannelManager; +import org.asynchttpclient.netty.channel.Channels; import org.asynchttpclient.netty.request.NettyRequestSender; import org.asynchttpclient.util.HttpConstants.ResponseStatusCodes; @@ -56,6 +59,26 @@ 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( + future, channel, future::touch, + bodyFullyRead -> finishUpdate(future, channel, !bodyFullyRead || !future.isKeepAlive())); + return handler.onResponseBodyStart(control) == State.ABORT; + } + + private static void ignoreInterimResponseTerminator(Channel channel, NettyResponseFuture future) { + // Bind the synthetic terminator to this exchange through the same callback mechanism used for deferred + // 100-continue bodies and response draining. The callback restores the future before a final response can be + // handled, and it cannot leave a channel marker behind for a later exchange. + Channels.setAttribute(channel, new OnLastHttpContentCallback(future) { + @Override + public void call() { + Channels.setAttribute(channel, future); + } + }); + } + private void handleHttpResponse(final HttpResponse response, final Channel channel, final NettyResponseFuture future, AsyncHandler handler) throws Exception { HttpRequest httpRequest = future.getNettyRequest().getHttpRequest(); if (logger.isDebugEnabled()) { @@ -66,12 +89,31 @@ private void handleHttpResponse(final HttpResponse response, final Channel chann NettyResponseStatus status = new NettyResponseStatus(future.getUri(), response, channel); HttpHeaders responseHeaders = response.headers(); + int statusCode = status.getStatusCode(); + + // RFC 9110 section 15.2: 1xx responses are interim, except 101 which switches protocols. Netty emits a + // synthetic LastHttpContent after each HTTP/1.1 interim response, so consume that terminator before accepting + // the final response. A deferred 100 Continue is the exception: its interceptor installs its own callback that + // uses the terminator to send the request body. + if (statusCode > 100 && statusCode < 200 + && statusCode != ResponseStatusCodes.SWITCHING_PROTOCOLS_101) { + ignoreInterimResponseTerminator(channel, future); + return; + } 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); } + } else if (statusCode == ResponseStatusCodes.CONTINUE_100 && Channels.getAttribute(channel) == future) { + // Continue100Interceptor replaces the future attribute with an OnLastHttpContentCallback only when this + // request actually deferred its body. If the attribute is still this future, the 100 was unsolicited and + // its synthetic terminator only needs to be consumed before waiting for the final response. + ignoreInterimResponseTerminator(channel, future); } } @@ -81,10 +123,15 @@ private void handleChunk(HttpContent chunk, final Channel channel, final NettyRe // Netty 4: the last chunk is not empty if (last) { + NettyResponseBodyControl.markBodyFullyRead(future); LastHttpContent lastChunk = (LastHttpContent) chunk; HttpHeaders trailingHeaders = lastChunk.trailingHeaders(); if (!trailingHeaders.isEmpty()) { abort = handler.onTrailingHeadersReceived(trailingHeaders) == State.ABORT; + // cancel() may have completed the future inline from the trailer callback. + if (future.isDone()) { + return; + } } } @@ -92,6 +139,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/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index c142bc62a..5e85e0a87 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -59,6 +59,7 @@ import org.asynchttpclient.filter.IOExceptionFilter; import org.asynchttpclient.handler.TransferCompletionHandler; import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.NettyResponseBodyControl; import org.asynchttpclient.netty.OnLastHttpContentCallback; import org.asynchttpclient.netty.SimpleFutureListener; import org.asynchttpclient.netty.channel.ChannelManager; @@ -1585,6 +1586,7 @@ public void replayRequest(final NettyResponseFuture future, FilterContext fc, future.setProxyServer(getProxyServer(config, newRequest)); future.setTargetRequest(newRequest); + NettyResponseBodyControl.complete(future); if (channel instanceof Http2StreamChannel) { Channels.setDiscard(channel); channelManager.closeChannel(channel); 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 18d3078b6..f9646e56f 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java @@ -17,14 +17,22 @@ import io.netty.util.Timeout; import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.NettyResponseBodyControl; import org.asynchttpclient.netty.request.NettyRequestSender; import org.asynchttpclient.util.StringBuilderPool; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.atomic.AtomicBoolean; import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; public class ReadTimeoutTimerTask extends TimeoutTimerTask implements Runnable { + private static final Logger LOGGER = LoggerFactory.getLogger(ReadTimeoutTimerTask.class); + private final long readTimeout; + private final AtomicBoolean indefiniteSuspensionWarningLogged = new AtomicBoolean(); ReadTimeoutTimerTask(NettyResponseFuture nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder, long readTimeout) { super(nettyResponseFuture, requestSender, timeoutsHolder); @@ -51,6 +59,13 @@ public void run(Timeout timeout) { return; } + if (NettyResponseBodyControl.isSuspended(nettyResponseFuture)) { + warnIfIndefinitelySuspended(); + done.set(false); + timeoutsHolder.startReadTimeout(this); + return; + } + long now = unpreciseMillisTime(); long currentReadTimeoutInstant = readTimeout + nettyResponseFuture.getLastTouch(); @@ -71,4 +86,13 @@ public void run(Timeout timeout) { timeoutsHolder.startReadTimeout(this); } } + + void warnIfIndefinitelySuspended() { + if (timeoutsHolder.isRequestTimeoutDisabled() + && indefiniteSuspensionWarningLogged.compareAndSet(false, true)) { + LOGGER.warn("Response body reads for {} remain suspended while the request timeout is disabled; " + + "the exchange retains its transport resources until it is resumed or canceled", + nettyResponseFuture.getUri()); + } + } } 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 995feabcd..d7fdf2839 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -169,6 +169,10 @@ long requestTimeoutMillisTime() { return requestTimeoutMillisTime; } + boolean isRequestTimeoutDisabled() { + return requestTimeoutTask == null; + } + /** * Moves this exchange's timeouts onto {@code executor}, the loop of the channel it turned out to run on. The * connect path arms the request timeout before there is a channel -- deliberately, since it bounds address 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 000000000..96b2c6d54 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/Http2ResponseBodyControlTest.java @@ -0,0 +1,446 @@ +/* + * 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.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; +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 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; + 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 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(false, 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()); + } + } + + @Test + public void suspensionCannotDeferBodylessResponse() throws Exception { + try (AsyncHttpClient client = http2Client()) { + RecordingHandler handler = new RecordingHandler(); + ListenableFuture request = client.prepareGet(url("/empty")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + assertSame(handler, request.get(5, SECONDS)); + control.suspend(); + control.resume(); + control.cancel(); + + assertEquals(0, 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 = writeFrames(ctx, FRAME_COUNT); + 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; + 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; + case "/empty": + ctx.writeAndFlush(new DefaultHttp2HeadersFrame( + new DefaultHttp2Headers().status("200"), true)); + 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 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 { + 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 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, false); + } + + private RecordingHandler(boolean suspendEveryPart) { + this(suspendEveryPart, false); + } + + private RecordingHandler(boolean suspendEveryPart, boolean cancelOnBodyPart) { + this.suspendEveryPart = suspendEveryPart; + 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) { + if (suspendEveryPart) { + responseBodyControl.suspend(); + } + items.add(new String(bytes, CharsetUtil.US_ASCII)); + firstBodyPart.complete(null); + } + 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 000000000..c182fd142 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/ResponseBodyControlTest.java @@ -0,0 +1,686 @@ +/* + * 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.DefaultLastHttpContent; +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.handler.ssl.SslContext; +import io.netty.handler.ssl.SslContextBuilder; +import io.netty.util.AttributeKey; +import io.netty.util.CharsetUtil; +import io.netty.util.ReferenceCountUtil; +import io.netty.util.concurrent.GlobalEventExecutor; +import io.netty.pkitesting.CertificateBuilder; +import io.netty.pkitesting.X509Bundle; +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 org.asynchttpclient.filter.FilterContext; +import org.asynchttpclient.filter.IOExceptionFilter; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static io.netty.handler.codec.http.HttpResponseStatus.EARLY_HINTS; +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.assertDoesNotThrow; +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 CompletableFuture firstReplayContext = new CompletableFuture<>(); + private final CompletableFuture secondReplayContext = new CompletableFuture<>(); + private final CountDownLatch cancelledConnectionClosed = new CountDownLatch(1); + private final AtomicInteger replayRequestCount = new AtomicInteger(); + + private NioEventLoopGroup serverGroup; + private Channel serverChannel; + private Channel tlsServerChannel; + private ChannelGroup serverChildChannels; + private SslContext tlsServerSslContext; + private int serverPort; + private int tlsServerPort; + + @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 (tlsServerChannel != null) { + tlsServerChannel.close().sync(); + } + if (serverGroup != null) { + serverGroup.shutdownGracefully(0, 100, MILLISECONDS).sync(); + } + ReferenceCountUtil.release(tlsServerSslContext); + } + + @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) throws IOException { + 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("1", replacement.getResponseBody()); + assertEquals(1, connectionCount.get(), "a fully read response can reuse its HTTP/1.1 connection"); + } + } + + @Test + public void cancellationFromTrailerCallbackSkipsTerminalBodyCallbackAndReusesConnection() throws Exception { + AtomicReference callbackControl = new AtomicReference<>(); + AtomicBoolean trailerSeen = new AtomicBoolean(); + AtomicInteger bodyPartCallsAfterTrailers = new AtomicInteger(); + 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 onTrailingHeadersReceived(io.netty.handler.codec.http.HttpHeaders headers) { + trailerSeen.set(true); + callbackControl.get().cancel(); + return State.CONTINUE; + } + + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws IOException { + if (trailerSeen.get()) { + bodyPartCallsAfterTrailers.incrementAndGet(); + } + return super.onBodyPartReceived(bodyPart); + } + }; + + assertSame(handler, client.prepareGet(url("/trailers")).execute(handler).get(5, SECONDS)); + assertTrue(trailerSeen.get()); + assertEquals(0, bodyPartCallsAfterTrailers.get(), + "cancellation from trailers must skip later terminal body callbacks"); + assertEquals(1, handler.completionCount.get()); + assertNull(handler.throwable.get()); + + Response replacement = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", replacement.getResponseBody()); + assertEquals(1, 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); + + assertSame(handler, request.get(5, SECONDS), "suspension cannot defer a bodyless response"); + control.suspend(); + control.resume(); + control.cancel(); + + assertTrue(handler.items.isEmpty()); + assertEquals(1, handler.completionCount.get()); + assertNull(handler.throwable.get()); + + Response pooled = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", pooled.getResponseBody()); + assertEquals(1, connectionCount.get()); + } + } + + @Test + public void earlyHintsDoNotStartOrCompleteTheResponseBody() throws Exception { + AtomicInteger statuses = new AtomicInteger(); + AtomicInteger headers = new AtomicInteger(); + AtomicInteger bodyStarts = new AtomicInteger(); + AtomicInteger finalStatus = new AtomicInteger(); + try (AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10)))) { + RecordingHandler handler = new RecordingHandler(false) { + @Override + public State onStatusReceived(HttpResponseStatus responseStatus) { + statuses.incrementAndGet(); + finalStatus.set(responseStatus.getStatusCode()); + return State.CONTINUE; + } + + @Override + public State onHeadersReceived(io.netty.handler.codec.http.HttpHeaders responseHeaders) { + headers.incrementAndGet(); + assertEquals("present", responseHeaders.get("final-header")); + assertNull(responseHeaders.get("link")); + return State.CONTINUE; + } + + @Override + public State onResponseBodyStart(ResponseBodyControl control) { + bodyStarts.incrementAndGet(); + return State.CONTINUE; + } + }; + + assertSame(handler, client.prepareGet(url("/early-hints")).execute(handler).get(5, SECONDS)); + assertEquals(1, statuses.get()); + assertEquals(OK.code(), finalStatus.get()); + assertEquals(1, headers.get()); + assertEquals(1, bodyStarts.get()); + assertEquals("final", handler.items.poll(5, SECONDS)); + assertEquals(1, handler.completionCount.get()); + assertNull(handler.throwable.get()); + + Response pooled = client.prepareGet(url("/pool")).execute().get(5, SECONDS); + assertEquals("1", pooled.getResponseBody()); + assertEquals(1, connectionCount.get()); + } + } + + @Test + public void ioExceptionReplayReplacesControlAndRestoresDrainingChannel() throws Exception { + startTlsServer(); + AtomicBoolean replay = new AtomicBoolean(); + IOExceptionFilter replayOnce = new IOExceptionFilter() { + @Override + public FilterContext filter(FilterContext ctx) { + if (ctx.getIOException() != null && "replay response".equals(ctx.getIOException().getMessage()) + && replay.compareAndSet(false, true)) { + return new FilterContext.FilterContextBuilder<>(ctx.getAsyncHandler(), ctx.getRequest()) + .replayRequest(true) + .build(); + } + return ctx; + } + }; + List clientChannels = new CopyOnWriteArrayList<>(); + AtomicInteger responseStarts = new AtomicInteger(); + AtomicBoolean failFirstBodyPart = new AtomicBoolean(true); + AtomicReference firstControl = new AtomicReference<>(); + CompletableFuture replacementControl = new CompletableFuture<>(); + + try (AsyncHttpClient client = asyncHttpClient(config() + .setUseInsecureTrustManager(true) + .setMaxRequestRetry(1) + .setRequestTimeout(Duration.ofSeconds(10)) + .addIOExceptionFilter(replayOnce) + .setHttpAdditionalChannelInitializer(clientChannels::add))) { + RecordingHandler handler = new RecordingHandler(false) { + @Override + public State onResponseBodyStart(ResponseBodyControl control) { + if (responseStarts.incrementAndGet() == 1) { + firstControl.set(control); + } else { + replacementControl.complete(control); + } + return State.CONTINUE; + } + + @Override + public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws IOException { + if (failFirstBodyPart.compareAndSet(true, false)) { + firstControl.get().suspend(); + throw new IOException("replay response"); + } + return super.onBodyPartReceived(bodyPart); + } + }; + + ListenableFuture request = client.prepareGet(httpsUrl("/replay")).execute(handler); + ChannelHandlerContext firstServer = firstReplayContext.get(5, SECONDS); + ResponseBodyControl replacement = replacementControl.get(5, SECONDS); + + Channel firstClient = clientChannels.get(0); + awaitEventLoop(firstClient); + assertTrue(firstClient.config().isAutoRead(), "replay must restore reads before draining the old response"); + + firstControl.get().suspend(); + firstControl.get().cancel(); + ChannelHandlerContext secondServer = secondReplayContext.get(5, SECONDS); + writeChunk(secondServer, "replayed"); + writeLast(secondServer); + replacement.resume(); + + assertSame(handler, request.get(5, SECONDS)); + assertEquals("replayed", handler.items.poll(5, SECONDS)); + assertEquals(2, responseStarts.get()); + assertNull(handler.throwable.get()); + + writeLast(firstServer); + } + } + + @Test + public void callsAfterClientShutdownDoNotThrow() throws Exception { + AsyncHttpClient client = asyncHttpClient(config().setRequestTimeout(Duration.ofSeconds(10))); + RecordingHandler handler = new RecordingHandler(false); + client.prepareGet(url("/cancel")).execute(handler); + ResponseBodyControl control = handler.control.get(5, SECONDS); + + client.close(); + + assertDoesNotThrow(control::suspend); + assertDoesNotThrow(control::resume); + assertDoesNotThrow(control::cancel); + } + + private String url(String path) { + return "http://localhost:" + serverPort + path; + } + + private String httpsUrl(String path) { + return "https://localhost:" + tlsServerPort + path; + } + + private void startTlsServer() throws Exception { + X509Bundle bundle = new CertificateBuilder() + .subject("CN=localhost") + .setIsCertificateAuthority(true) + .buildSelfSigned(); + tlsServerSslContext = SslContextBuilder.forServer(bundle.toKeyManagerFactory()).build(); + tlsServerChannel = 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(tlsServerSslContext.newHandler(channel.alloc())) + .addLast(new HttpServerCodec()) + .addLast(new HttpObjectAggregator(1024)) + .addLast(new StreamingServerHandler()); + } + }) + .bind(0) + .sync() + .channel(); + tlsServerPort = ((InetSocketAddress) tlsServerChannel.localAddress()).getPort(); + } + + 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; + case "/replay": + writeStreamingHeaders(ctx); + if (replayRequestCount.incrementAndGet() == 1) { + firstReplayContext.complete(ctx); + ctx.writeAndFlush(new DefaultHttpContent( + Unpooled.copiedBuffer("first", CharsetUtil.US_ASCII))); + } else { + secondReplayContext.complete(ctx); + } + break; + case "/trailers": + HttpResponse trailerResponse = streamingResponse(); + trailerResponse.headers().set("trailer", "test-trailer"); + ctx.write(trailerResponse); + DefaultLastHttpContent last = new DefaultLastHttpContent( + Unpooled.copiedBuffer("last", CharsetUtil.US_ASCII)); + last.trailingHeaders().set("test-trailer", "present"); + ctx.writeAndFlush(last); + break; + case "/early-hints": + HttpResponse earlyHints = new DefaultHttpResponse(HTTP_1_1, EARLY_HINTS); + earlyHints.headers().set("link", "; rel=preload; as=style"); + ctx.write(earlyHints); + ctx.write(LastHttpContent.EMPTY_LAST_CONTENT); + ByteBuf finalContent = Unpooled.copiedBuffer("final", CharsetUtil.US_ASCII); + DefaultFullHttpResponse finalResponse = new DefaultFullHttpResponse(HTTP_1_1, OK, finalContent); + finalResponse.headers().set("final-header", "present"); + HttpUtil.setContentLength(finalResponse, finalContent.readableBytes()); + HttpUtil.setKeepAlive(finalResponse, true); + ctx.writeAndFlush(finalResponse); + 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) { + ctx.writeAndFlush(streamingResponse()); + } + + private HttpResponse streamingResponse() { + HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + HttpUtil.setTransferEncodingChunked(response, true); + HttpUtil.setKeepAlive(response, true); + return 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) throws IOException { + 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/channel/SuspensionAwareHttp2LocalFlowControllerTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/SuspensionAwareHttp2LocalFlowControllerTest.java new file mode 100644 index 000000000..317ce0d34 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/SuspensionAwareHttp2LocalFlowControllerTest.java @@ -0,0 +1,180 @@ +/* + * 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.channel; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import io.netty.handler.codec.http2.DefaultHttp2Connection; +import io.netty.handler.codec.http2.Http2Connection; +import io.netty.handler.codec.http2.Http2Exception; +import io.netty.handler.codec.http2.Http2FrameWriter; +import io.netty.handler.codec.http2.Http2Stream; +import io.netty.util.concurrent.EventExecutor; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID; +import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_WINDOW_SIZE; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class SuspensionAwareHttp2LocalFlowControllerTest { + + private static final int STREAM_ID = 1; + private static final int SECOND_STREAM_ID = 3; + private static final int WINDOW_UPDATE_SIZE = DEFAULT_WINDOW_SIZE / 2 + 1; + + private final Http2FrameWriter frameWriter = mock(Http2FrameWriter.class); + private final ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + private final ChannelPromise promise = mock(ChannelPromise.class); + private final EventExecutor executor = mock(EventExecutor.class); + + private Http2Connection connection; + private SuspensionAwareHttp2LocalFlowController controller; + + @BeforeEach + public void setUp() throws Http2Exception { + reset(frameWriter, ctx, promise, executor); + when(ctx.newPromise()).thenReturn(promise); + when(ctx.executor()).thenReturn(executor); + when(executor.inEventLoop()).thenReturn(true); + + connection = new DefaultHttp2Connection(false); + controller = new SuspensionAwareHttp2LocalFlowController(connection).frameWriter(frameWriter); + connection.local().flowController(controller); + connection.local().createStream(STREAM_ID, false); + connection.local().createStream(SECOND_STREAM_ID, false); + controller.channelHandlerContext(ctx); + } + + @Test + public void retainsNormalConnectionAccountingWithoutSuspension() throws Http2Exception { + receive(STREAM_ID, WINDOW_UPDATE_SIZE); + + assertEquals(WINDOW_UPDATE_SIZE, controller.unconsumedBytes(connection.connectionStream())); + assertEquals(0, controller.autoConsumedConnectionBytes()); + verifyNoWindowUpdate(); + + assertTrue(controller.consumeBytes(stream(STREAM_ID), WINDOW_UPDATE_SIZE)); + verifyWindowUpdate(CONNECTION_STREAM_ID, WINDOW_UPDATE_SIZE); + verifyWindowUpdate(STREAM_ID, WINDOW_UPDATE_SIZE); + } + + @Test + public void refillsOnlyConnectionCreditWhileSuspended() throws Http2Exception { + controller.suspendResponse(); + receive(STREAM_ID, WINDOW_UPDATE_SIZE); + + assertEquals(0, controller.unconsumedBytes(connection.connectionStream())); + assertEquals(WINDOW_UPDATE_SIZE, controller.autoConsumedConnectionBytes()); + assertEquals(WINDOW_UPDATE_SIZE, controller.unconsumedBytes(stream(STREAM_ID))); + verifyWindowUpdate(CONNECTION_STREAM_ID, WINDOW_UPDATE_SIZE); + verifyNoWindowUpdate(STREAM_ID); + + assertTrue(controller.consumeBytes(stream(STREAM_ID), WINDOW_UPDATE_SIZE)); + assertEquals(0, controller.autoConsumedConnectionBytes()); + verifyWindowUpdate(STREAM_ID, WINDOW_UPDATE_SIZE); + } + + @Test + public void refillsCreditReceivedBeforeSuspendCallback() throws Http2Exception { + receive(STREAM_ID, WINDOW_UPDATE_SIZE); + verifyNoWindowUpdate(); + + controller.suspendResponse(); + + assertEquals(0, controller.unconsumedBytes(connection.connectionStream())); + assertEquals(WINDOW_UPDATE_SIZE, controller.autoConsumedConnectionBytes()); + verifyWindowUpdate(CONNECTION_STREAM_ID, WINDOW_UPDATE_SIZE); + verifyNoWindowUpdate(STREAM_ID); + } + + @Test + public void continuesRefillingUntilLastSuspendedResponseResumes() throws Http2Exception { + controller.suspendResponse(); + controller.suspendResponse(); + controller.resumeResponse(); + assertTrue(controller.hasSuspendedResponse()); + + receive(STREAM_ID, WINDOW_UPDATE_SIZE); + verifyWindowUpdate(CONNECTION_STREAM_ID, WINDOW_UPDATE_SIZE); + controller.consumeBytes(stream(STREAM_ID), WINDOW_UPDATE_SIZE); + + controller.resumeResponse(); + assertFalse(controller.hasSuspendedResponse()); + reset(frameWriter); + + receive(SECOND_STREAM_ID, WINDOW_UPDATE_SIZE); + verifyNoWindowUpdate(); + controller.consumeBytes(stream(SECOND_STREAM_ID), WINDOW_UPDATE_SIZE); + verifyWindowUpdate(CONNECTION_STREAM_ID, WINDOW_UPDATE_SIZE); + verifyWindowUpdate(SECOND_STREAM_ID, WINDOW_UPDATE_SIZE); + } + + @Test + public void doesNotReturnConnectionCreditTwiceAfterResume() throws Http2Exception { + controller.suspendResponse(); + receive(STREAM_ID, WINDOW_UPDATE_SIZE); + controller.resumeResponse(); + receive(SECOND_STREAM_ID, WINDOW_UPDATE_SIZE); + reset(frameWriter); + + controller.consumeBytes(stream(SECOND_STREAM_ID), WINDOW_UPDATE_SIZE); + verifyNoWindowUpdate(CONNECTION_STREAM_ID); + verifyWindowUpdate(SECOND_STREAM_ID, WINDOW_UPDATE_SIZE); + + controller.consumeBytes(stream(STREAM_ID), WINDOW_UPDATE_SIZE); + assertEquals(0, controller.autoConsumedConnectionBytes()); + verifyWindowUpdate(CONNECTION_STREAM_ID, WINDOW_UPDATE_SIZE); + verifyWindowUpdate(STREAM_ID, WINDOW_UPDATE_SIZE); + } + + private void receive(int streamId, int size) throws Http2Exception { + ByteBuf data = Unpooled.buffer(size).writerIndex(size); + try { + controller.receiveFlowControlledFrame(stream(streamId), data, 0, false); + } finally { + data.release(); + } + } + + private Http2Stream stream(int streamId) { + return connection.stream(streamId); + } + + private void verifyWindowUpdate(int streamId, int increment) { + verify(frameWriter).writeWindowUpdate(ctx, streamId, increment, promise); + } + + private void verifyNoWindowUpdate(int streamId) { + verify(frameWriter, never()).writeWindowUpdate(eq(ctx), eq(streamId), anyInt(), eq(promise)); + } + + private void verifyNoWindowUpdate() { + verify(frameWriter, never()).writeWindowUpdate(any(), anyInt(), anyInt(), any()); + } +} 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 2a5f5e205..85c363d27 100644 --- a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutTimerTaskTest.java @@ -15,16 +15,26 @@ */ package org.asynchttpclient.netty.timeout; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClientConfig; import org.asynchttpclient.DefaultAsyncHttpClientConfig; import org.asynchttpclient.Request; import org.asynchttpclient.RequestBuilder; import org.asynchttpclient.channel.ChannelPoolPartitioning; import org.asynchttpclient.netty.NettyResponseFuture; import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.List; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; public class TimeoutTimerTaskTest { @@ -84,4 +94,62 @@ 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()); + } + + @Test + public void indefiniteSuspensionLogsOneWarning() { + 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); + AsyncHttpClientConfig config = new DefaultAsyncHttpClientConfig.Builder() + .setReadTimeout(Duration.ofSeconds(1)) + .setRequestTimeout(Duration.ofMillis(-1)) + .build(); + TimeoutsHolder timeoutsHolder = new TimeoutsHolder(null, future, null, config, null); + ReadTimeoutTimerTask task = new ReadTimeoutTimerTask(future, null, timeoutsHolder, 1_000); + + Logger logger = (Logger) LoggerFactory.getLogger(ReadTimeoutTimerTask.class); + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + try { + task.warnIfIndefinitelySuspended(); + task.warnIfIndefinitelySuspended(); + + List warnings = appender.list.stream() + .filter(event -> event.getLevel() == Level.WARN) + .collect(java.util.stream.Collectors.toList()); + assertEquals(1, warnings.size()); + assertTrue(warnings.get(0).getFormattedMessage().contains("request timeout is disabled")); + } finally { + logger.detachAppender(appender); + appender.stop(); + } + } }