From 7b2f76ef9247dff1c12e89585b879b12d0194d39 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:10:15 +0700 Subject: [PATCH 1/3] fix(netty): fail fast when InputStream body cannot be reset HTTP/1 path silently returned after a warn when a consumed non-resettable InputStream body was reused, leaving a half-sent request that hung until timeout. Throw IOException so sendHttpRequest aborts the future (matches existing HTTP/2 behavior). Composer 2.5 on behalf of arimu1 Fixes #1973 Co-Authored-By: Composer 2.5 --- .../request/body/NettyInputStreamBody.java | 10 ++- .../body/NettyInputStreamBodyTest.java | 72 +++++++++++++++++++ 2 files changed, 79 insertions(+), 3 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java index 32dbdc0fe..28132e233 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java @@ -65,8 +65,12 @@ public void write(Channel channel, NettyResponseFuture future) throws IOExcep if (is.markSupported()) { is.reset(); } else { - LOGGER.warn("Stream has already been consumed and cannot be reset"); - return; + // The request headers were already written (sendHttpRequest), so silently returning would + // leave the request half-sent with no terminating LastHttpContent — the request would then + // hang until it times out (the Issue #1973 silent-timeout class). A non-resettable + // InputStream cannot be replayed (retry / redirect / auth), so fail explicitly: the caller + // (sendHttpRequest) aborts the future on the IOException. + throw new IOException("HTTP/1 request body InputStream already consumed and cannot be reset for a retry"); } } else { future.setStreamConsumed(true); @@ -93,7 +97,7 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture future } else { // The HEADERS frame was already written with endStream=false (sendHttp2Frames), so silently // returning would leave the stream half-open with no terminating DATA frame — the request - // would then hang until it times out (the Issue #2160 silent-timeout class). A non-resettable + // would then hang until it times out (the Issue #1973 silent-timeout class). A non-resettable // InputStream cannot be replayed (retry / redirect / auth), so fail the stream explicitly: // the caller (openHttp2Stream / sendHttp2RequestBody) aborts this single stream on the // IOException, leaving sibling multiplexed streams untouched. diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java new file mode 100644 index 000000000..67940e245 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java @@ -0,0 +1,72 @@ +/* + * 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.request.body; + +import io.netty.channel.embedded.EmbeddedChannel; +import org.asynchttpclient.AsyncHandler; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.InputStream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; + +/** + * Regression guard for Issue #1973: a non-resettable {@link InputStream} body that has already been + * consumed must fail the request immediately on retry instead of hanging until the request timeout. + */ +public class NettyInputStreamBodyTest { + + @Test + public void http1WriteFailsWhenStreamAlreadyConsumedAndNotResettable() { + NettyInputStreamBody body = new NettyInputStreamBody(new NonResettableInputStream(new byte[] {1, 2, 3})); + NettyResponseFuture future = new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 0, null, null, null); + future.setStreamConsumed(true); + + EmbeddedChannel channel = new EmbeddedChannel(); + try { + IOException ex = assertThrows(IOException.class, () -> body.write(channel, future)); + assertEquals( + "HTTP/1 request body InputStream already consumed and cannot be reset for a retry", + ex.getMessage()); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static final class NonResettableInputStream extends InputStream { + + private final byte[] data; + private int index; + + NonResettableInputStream(byte[] data) { + this.data = data; + } + + @Override + public int read() { + return index < data.length ? data[index++] & 0xFF : -1; + } + + @Override + public boolean markSupported() { + return false; + } + } +} From 992b0a0c90636073994264f58404725c0a1ea316 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:10:15 +0700 Subject: [PATCH 2/3] fix(netty): drop unused LOGGER after InputStream fail-fast Composer 2.5 on behalf of arimu1 Co-Authored-By: Composer 2.5 --- .../netty/request/body/NettyInputStreamBody.java | 4 ---- 1 file changed, 4 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java index 28132e233..878cf434d 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java @@ -24,8 +24,6 @@ import io.netty.handler.stream.ChunkedStream; import org.asynchttpclient.netty.NettyResponseFuture; import org.asynchttpclient.netty.request.WriteProgressListener; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.InputStream; @@ -34,8 +32,6 @@ public class NettyInputStreamBody implements NettyBody { - private static final Logger LOGGER = LoggerFactory.getLogger(NettyInputStreamBody.class); - private final InputStream inputStream; private final long contentLength; From da62c2fd1d3f0b66e1f6f93dcd98eb44d01df54e Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Sun, 16 Aug 2026 07:30:18 +0700 Subject: [PATCH 3/3] fix(netty): fail-fast consumed InputStream retry Unsolicited HTTP/1 100 Continue was replaying bodies that were already sent. Guard with bodyWasDeferred like HTTP/2. Treat a failed reset() on a closed markable stream as unreplayable so retries abort the future instead of hanging or throwing Stream closed. Comments now cite NettyRequestSender.writeRequest. Addresses review on #2312. Composer 2.5 on behalf of arimu1 Co-Authored-By: Composer 2.5 --- .../intercept/Continue100Interceptor.java | 7 +- .../request/body/NettyInputStreamBody.java | 47 +++-- .../intercept/Continue100InterceptorTest.java | 96 +++++++++ ...ttyRequestSenderConsumedBodyRetryTest.java | 183 ++++++++++++++++++ .../body/NettyInputStreamBodyTest.java | 90 ++++++--- 5 files changed, 377 insertions(+), 46 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/netty/handler/intercept/Continue100InterceptorTest.java create mode 100644 client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderConsumedBodyRetryTest.java diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java index fec5daa6e..a17f74e31 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java @@ -56,8 +56,11 @@ public boolean exitAfterHandling100(final Channel channel, final NettyResponseFu requestSender.abort(channel, future, e); } } - } else { - // HTTP/1.1: wait for LastHttpContent before sending the body + } else if (bodyWasDeferred) { + // HTTP/1.1: wait for LastHttpContent before sending the deferred body. + // Unsolicited 100 when the body was already sent must not replay it + // (RFC 9110 15.2.1). The same guard also avoids a second LastHttpContent + // on a keep-alive socket whose first LastHttpContent was already flushed. Channels.setAttribute(channel, new OnLastHttpContentCallback(future) { @Override public void call() { diff --git a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java index 878cf434d..8c69000b1 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/body/NettyInputStreamBody.java @@ -58,16 +58,11 @@ public void write(Channel channel, NettyResponseFuture future) throws IOExcep final InputStream is = inputStream; if (future.isStreamConsumed()) { - if (is.markSupported()) { - is.reset(); - } else { - // The request headers were already written (sendHttpRequest), so silently returning would - // leave the request half-sent with no terminating LastHttpContent — the request would then - // hang until it times out (the Issue #1973 silent-timeout class). A non-resettable - // InputStream cannot be replayed (retry / redirect / auth), so fail explicitly: the caller - // (sendHttpRequest) aborts the future on the IOException. - throw new IOException("HTTP/1 request body InputStream already consumed and cannot be reset for a retry"); - } + // Silently returning would leave NettyRequestSender.writeRequest without completing + // the future, so the request hangs until timeout (Issue #1973). Headers used + // channel.write with no flush, so the peer has seen nothing; the skipped flush is + // writeAndFlush of LastHttpContent. writeRequest's catch aborts the future. + replayConsumedStream(is, "HTTP/1 request body InputStream already consumed and cannot be reset for a retry"); } else { future.setStreamConsumed(true); } @@ -88,17 +83,13 @@ public void writeHttp2(Http2StreamChannel channel, NettyResponseFuture future final InputStream is = inputStream; if (future.isStreamConsumed()) { - if (is.markSupported()) { - is.reset(); - } else { - // The HEADERS frame was already written with endStream=false (sendHttp2Frames), so silently - // returning would leave the stream half-open with no terminating DATA frame — the request - // would then hang until it times out (the Issue #1973 silent-timeout class). A non-resettable - // InputStream cannot be replayed (retry / redirect / auth), so fail the stream explicitly: - // the caller (openHttp2Stream / sendHttp2RequestBody) aborts this single stream on the - // IOException, leaving sibling multiplexed streams untouched. - throw new IOException("HTTP/2 request body InputStream already consumed and cannot be reset for a retry"); - } + // The HEADERS frame was already written with endStream=false (sendHttp2Frames), so silently + // returning would leave the stream half-open with no terminating DATA frame — the request + // would then hang until it times out (the Issue #2160 silent-timeout class). A non-resettable + // InputStream cannot be replayed (retry / redirect / auth), so fail the stream explicitly: + // the caller (openHttp2Stream / sendHttp2RequestBody) aborts this single stream on the + // IOException, leaving sibling multiplexed streams untouched. + replayConsumedStream(is, "HTTP/2 request body InputStream already consumed and cannot be reset for a retry"); } else { future.setStreamConsumed(true); } @@ -152,4 +143,18 @@ public void close() { closeSilently(is); } } + + // markSupported is insufficient: WriteProgressListener closes the stream after the first write, + // so a BufferedInputStream still reports mark support and reset() fails with "Stream closed". + private static void replayConsumedStream(InputStream is, String message) throws IOException { + if (is.markSupported()) { + try { + is.reset(); + return; + } catch (IOException e) { + throw new IOException(message, e); + } + } + throw new IOException(message); + } } diff --git a/client/src/test/java/org/asynchttpclient/netty/handler/intercept/Continue100InterceptorTest.java b/client/src/test/java/org/asynchttpclient/netty/handler/intercept/Continue100InterceptorTest.java new file mode 100644 index 000000000..3acd3e08e --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/handler/intercept/Continue100InterceptorTest.java @@ -0,0 +1,96 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.handler.intercept; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AsyncHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.netty.NettyResponseFuture; +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.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +/** + * HTTP/1 must replay a body on 100 Continue only when that body was deferred for Expect: 100-continue. + */ +public class Continue100InterceptorTest { + + private ChannelManager channelManager; + private Timer timer; + private Continue100Interceptor interceptor; + + @BeforeEach + public void setUp() { + AsyncHttpClientConfig cfg = config().build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(cfg, timer); + interceptor = new Continue100Interceptor(new NettyRequestSender(cfg, channelManager, timer, null)); + } + + @AfterEach + public void tearDown() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + @Test + public void http1Unsolicited100DoesNotReplayABodyThatWasAlreadySent() { + NettyResponseFuture future = newFuture(); + future.setDontWriteBodyBecauseExpectContinue(false); + + EmbeddedChannel channel = new EmbeddedChannel(); + try { + assertTrue(interceptor.exitAfterHandling100(channel, future)); + assertFalse(Channels.getAttribute(channel) instanceof OnLastHttpContentCallback); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void http1Deferred100SchedulesTheBodyWriteAfterLastHttpContent() { + NettyResponseFuture future = newFuture(); + future.setDontWriteBodyBecauseExpectContinue(true); + + EmbeddedChannel channel = new EmbeddedChannel(); + try { + assertTrue(interceptor.exitAfterHandling100(channel, future)); + assertFalse(future.isDontWriteBodyBecauseExpectContinue()); + assertTrue(Channels.getAttribute(channel) instanceof OnLastHttpContentCallback); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static NettyResponseFuture newFuture() { + return new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 0, null, null, null); + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderConsumedBodyRetryTest.java b/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderConsumedBodyRetryTest.java new file mode 100644 index 000000000..3d5b13390 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/request/NettyRequestSenderConsumedBodyRetryTest.java @@ -0,0 +1,183 @@ +/* + * 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.request; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.stream.ChunkedWriteHandler; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.asynchttpclient.AbstractBasicTest; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClient; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.filter.FilterContext; +import org.asynchttpclient.filter.ResponseFilter; +import org.asynchttpclient.netty.NettyResponseFuture; +import org.asynchttpclient.netty.channel.ChannelManager; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.BufferedInputStream; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.time.Duration; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.asynchttpclient.Dsl.asyncHttpClient; +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Issue #1973: a consumed InputStream body on retry must complete the future promptly through + * {@link NettyRequestSender#writeRequest}, not hang until the request timeout. + */ +public class NettyRequestSenderConsumedBodyRetryTest extends AbstractBasicTest { + + private AsyncHttpClientConfig senderConfig; + private ChannelManager channelManager; + private NettyRequestSender sender; + private Timer timer; + + @BeforeEach + public void setUpSender() { + senderConfig = config().build(); + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(senderConfig, timer); + sender = new NettyRequestSender(senderConfig, channelManager, timer, null); + } + + @AfterEach + public void tearDownSender() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + @Test + public void writeRequestAbortsWhenConsumedStreamCannotReset() throws Exception { + Request request = new RequestBuilder("POST") + .setUrl("http://example.com/") + .setBody(InputStream.nullInputStream()) + .build(); + NettyRequestFactory factory = new NettyRequestFactory(senderConfig); + NettyResponseFuture future = newFuture(request, factory.newNettyRequest(request, false, null, null, null)); + + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler()); + try { + sender.writeRequest(future, channel); + assertFalse(future.isDone(), "first write of an unconsumed stream must not abort"); + + future.setNettyRequest(factory.newNettyRequest(request, false, null, null, null)); + sender.writeRequest(future, channel); + + ExecutionException thrown = assertThrows(ExecutionException.class, + () -> future.get(2, TimeUnit.SECONDS)); + assertInstanceOf(IOException.class, thrown.getCause()); + assertNotEquals("Stream closed", thrown.getCause().getMessage()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void writeRequestAbortsWhenMarkSupportedResetFails() throws Exception { + BufferedInputStream is = new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3})); + Request request = new RequestBuilder("POST") + .setUrl("http://example.com/") + .setBody(is) + .build(); + NettyRequestFactory factory = new NettyRequestFactory(senderConfig); + NettyResponseFuture future = newFuture(request, factory.newNettyRequest(request, false, null, null, null)); + + EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler()); + try { + sender.writeRequest(future, channel); + channel.runPendingTasks(); + assertFalse(future.isDone(), "first write of an unconsumed stream must not abort"); + assertThrows(IOException.class, is::read); + + future.setNettyRequest(factory.newNettyRequest(request, false, null, null, null)); + sender.writeRequest(future, channel); + + ExecutionException thrown = assertThrows(ExecutionException.class, + () -> future.get(2, TimeUnit.SECONDS)); + assertInstanceOf(IOException.class, thrown.getCause()); + assertNotEquals("Stream closed", thrown.getCause().getMessage()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void replayOfNonResettableInputStreamFailsTheFuturePromptly() throws Exception { + assertReplayFailsPromptly(InputStream.nullInputStream()); + } + + @Test + public void replayOfClosedBufferedInputStreamFailsTheFuturePromptly() throws Exception { + assertReplayFailsPromptly(new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3}))); + } + + private void assertReplayFailsPromptly(InputStream body) throws Exception { + AtomicBoolean replay = new AtomicBoolean(true); + ResponseFilter replayOnce = new ResponseFilter() { + @Override + public FilterContext filter(FilterContext ctx) { + if (replay.getAndSet(false)) { + return new FilterContext.FilterContextBuilder(ctx.getAsyncHandler(), ctx.getRequest()) + .replayRequest(true) + .build(); + } + return ctx; + } + }; + + try (AsyncHttpClient client = asyncHttpClient(config() + .addResponseFilter(replayOnce) + .setRequestTimeout(Duration.ofSeconds(30)))) { + ExecutionException thrown = assertThrows(ExecutionException.class, () -> + client.preparePost(getTargetUrl()) + .setBody(body) + .execute() + .get(2, TimeUnit.SECONDS)); + assertInstanceOf(IOException.class, thrown.getCause()); + assertNotEquals("Stream closed", thrown.getCause().getMessage()); + } + } + + private static NettyResponseFuture newFuture(Request request, NettyRequest nettyRequest) { + return new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }, nettyRequest, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java b/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java index 67940e245..9ae386ef0 100644 --- a/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/request/body/NettyInputStreamBodyTest.java @@ -16,57 +16,101 @@ package org.asynchttpclient.netty.request.body; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http2.Http2StreamChannel; import org.asynchttpclient.AsyncHandler; import org.asynchttpclient.netty.NettyResponseFuture; import org.junit.jupiter.api.Test; +import java.io.BufferedInputStream; +import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; -import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; /** - * Regression guard for Issue #1973: a non-resettable {@link InputStream} body that has already been - * consumed must fail the request immediately on retry instead of hanging until the request timeout. + * Regression guard for Issue #1973: a consumed InputStream body must fail the request immediately on + * retry instead of hanging until the request timeout. */ public class NettyInputStreamBodyTest { @Test - public void http1WriteFailsWhenStreamAlreadyConsumedAndNotResettable() { - NettyInputStreamBody body = new NettyInputStreamBody(new NonResettableInputStream(new byte[] {1, 2, 3})); - NettyResponseFuture future = new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 0, null, null, null); - future.setStreamConsumed(true); + public void http1WriteFailsWhenStreamAlreadyConsumedAndNotResettable() throws IOException { + NettyInputStreamBody body = new NettyInputStreamBody(InputStream.nullInputStream()); + NettyResponseFuture future = newFuture(true); EmbeddedChannel channel = new EmbeddedChannel(); try { IOException ex = assertThrows(IOException.class, () -> body.write(channel, future)); - assertEquals( - "HTTP/1 request body InputStream already consumed and cannot be reset for a retry", - ex.getMessage()); + assertNotEquals("Stream closed", ex.getMessage()); } finally { channel.finishAndReleaseAll(); } } - private static final class NonResettableInputStream extends InputStream { - - private final byte[] data; - private int index; + @Test + public void http1WriteFailsWhenConsumedMarkSupportedStreamCannotReset() throws IOException { + BufferedInputStream is = new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3})); + is.close(); + NettyInputStreamBody body = new NettyInputStreamBody(is); + NettyResponseFuture future = newFuture(true); - NonResettableInputStream(byte[] data) { - this.data = data; + EmbeddedChannel channel = new EmbeddedChannel(); + try { + IOException ex = assertThrows(IOException.class, () -> body.write(channel, future)); + assertNotEquals("Stream closed", ex.getMessage()); + assertTrue(ex.getCause() instanceof IOException); + } finally { + channel.finishAndReleaseAll(); } + } - @Override - public int read() { - return index < data.length ? data[index++] & 0xFF : -1; - } + @Test + public void http1FirstWriteClosesTheStreamAndRetryThenFailsFast() throws IOException { + BufferedInputStream is = new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3})); + NettyInputStreamBody body = new NettyInputStreamBody(is); + NettyResponseFuture future = newFuture(false); + + EmbeddedChannel channel = new EmbeddedChannel(); + try { + body.write(channel, future); + channel.runPendingTasks(); + assertThrows(IOException.class, is::read); - @Override - public boolean markSupported() { - return false; + IOException ex = assertThrows(IOException.class, () -> body.write(channel, future)); + assertNotEquals("Stream closed", ex.getMessage()); + } finally { + channel.finishAndReleaseAll(); } } + + @Test + public void http2WriteFailsWhenStreamAlreadyConsumedAndNotResettable() { + NettyInputStreamBody body = new NettyInputStreamBody(InputStream.nullInputStream()); + NettyResponseFuture future = newFuture(true); + + IOException ex = assertThrows(IOException.class, () -> body.writeHttp2(mock(Http2StreamChannel.class), future)); + assertNotEquals("Stream closed", ex.getMessage()); + } + + @Test + public void http2WriteFailsWhenConsumedMarkSupportedStreamCannotReset() throws IOException { + BufferedInputStream is = new BufferedInputStream(new ByteArrayInputStream(new byte[]{1, 2, 3})); + is.close(); + NettyInputStreamBody body = new NettyInputStreamBody(is); + NettyResponseFuture future = newFuture(true); + + IOException ex = assertThrows(IOException.class, () -> body.writeHttp2(mock(Http2StreamChannel.class), future)); + assertNotEquals("Stream closed", ex.getMessage()); + assertTrue(ex.getCause() instanceof IOException); + } + + private static NettyResponseFuture newFuture(boolean streamConsumed) { + NettyResponseFuture future = new NettyResponseFuture<>(null, mock(AsyncHandler.class), null, 0, null, null, null); + future.setStreamConsumed(streamConsumed); + return future; + } }