From 5f49e19f84651a57bee249373fa8af99d35a156b Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:31:13 +0200 Subject: [PATCH 1/2] Read InputStream bodies straight into the buffer, no staging array MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit InputStreamBodyGenerator's Body.transferTo allocated a fresh byte[] (sized to the target's writable region) on every call, read the stream into it, then copied those bytes into the target ByteBuf — a per-chunk allocation plus a redundant copy. Read directly into the target via ByteBuf.writeBytes(InputStream, int) instead (the same approach FileLikeMultipartPart uses), dropping both the staging array and the copy; the per-instance chunk field is gone. Behaviour is unchanged: it writes the bytes read this call and returns CONTINUE while data remains, STOP at EOF or on an I/O error (logged, as before). The '- 10' writable margin is preserved. Reachability note: AsyncHttpClient's own request path routes an InputStreamBodyGenerator to NettyInputStreamBody (NettyRequestFactory), so this Body is reached only through the public InputStreamBodyGenerator.createBody() API (external/custom callers); the change improves that path rather than pruning a public type. Adds InputStreamBodyGeneratorTest covering byte-for-byte transfer across multiple reads, a single read draining a small stream, and immediate STOP on an empty stream. No public API change. Fixes finding #8. --- .../generator/InputStreamBodyGenerator.java | 21 ++-- .../InputStreamBodyGeneratorTest.java | 95 +++++++++++++++++++ 2 files changed, 102 insertions(+), 14 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java diff --git a/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java b/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java index 1f602dae40..eecbc75ce1 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java +++ b/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java @@ -58,7 +58,6 @@ private static class InputStreamBody implements Body { private final InputStream inputStream; private final long contentLength; - private byte[] chunk; private InputStreamBody(InputStream inputStream, long contentLength) { this.inputStream = inputStream; @@ -72,23 +71,17 @@ public long getContentLength() { @Override public BodyState transferTo(ByteBuf target) { - - // To be safe. - chunk = new byte[target.writableBytes() - 10]; - - int read = -1; - boolean write = false; + // Read straight from the stream into the target buffer: no per-call staging byte[] and no extra + // copy (ByteBuf.writeBytes(InputStream, int) fills the buffer directly, like FileLikeMultipartPart). + // The "- 10" margin preserves the previous behaviour of never fully filling the writable region. + int read; try { - read = inputStream.read(chunk); + read = target.writeBytes(inputStream, target.writableBytes() - 10); } catch (IOException ex) { LOGGER.warn("Unable to read", ex); + return BodyState.STOP; } - - if (read > 0) { - target.writeBytes(chunk, 0, read); - write = true; - } - return write ? BodyState.CONTINUE : BodyState.STOP; + return read > 0 ? BodyState.CONTINUE : BodyState.STOP; } @Override diff --git a/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java b/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java new file mode 100644 index 0000000000..6c9140840f --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java @@ -0,0 +1,95 @@ +/* + * 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.request.body.generator; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.asynchttpclient.request.body.Body; +import org.asynchttpclient.request.body.Body.BodyState; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Random; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Covers {@link InputStreamBodyGenerator}'s {@link Body#transferTo(ByteBuf)}, which now reads straight from + * the stream into the target buffer (no per-call staging {@code byte[]} and no extra copy). The whole stream + * must still be transferred byte-for-byte, CONTINUE while data remains and STOP at EOF. + */ +public class InputStreamBodyGeneratorTest { + + private static final int CHUNK_SIZE = 1024 * 8; + + @Test + public void streamsAllBytesAcrossMultipleReads() throws IOException { + final byte[] src = new byte[3 * CHUNK_SIZE + 42]; + new Random().nextBytes(src); + + Body body = new InputStreamBodyGenerator(new ByteArrayInputStream(src)).createBody(); + ByteBuf chunkBuffer = Unpooled.buffer(CHUNK_SIZE); + ByteArrayOutputStream collected = new ByteArrayOutputStream(); + try { + BodyState state; + while ((state = body.transferTo(chunkBuffer)) != BodyState.STOP) { + assertEquals(BodyState.CONTINUE, state, "a stream with data left must report CONTINUE"); + byte[] b = new byte[chunkBuffer.readableBytes()]; + chunkBuffer.readBytes(b); + collected.write(b); + chunkBuffer.clear(); + } + assertArrayEquals(src, collected.toByteArray(), "the whole stream must be transferred unchanged"); + } finally { + chunkBuffer.release(); + body.close(); + } + } + + @Test + public void singleReadDrainsASmallStream() throws IOException { + final byte[] src = new byte[CHUNK_SIZE - 100]; // fits under writableBytes - 10, so one read drains it + new Random().nextBytes(src); + + Body body = new InputStreamBodyGenerator(new ByteArrayInputStream(src)).createBody(); + ByteBuf chunkBuffer = Unpooled.buffer(CHUNK_SIZE); + try { + assertEquals(BodyState.CONTINUE, body.transferTo(chunkBuffer)); + assertEquals(src.length, chunkBuffer.readableBytes(), "one read should drain a small stream"); + chunkBuffer.clear(); + assertEquals(BodyState.STOP, body.transferTo(chunkBuffer), "body at EOF"); + } finally { + chunkBuffer.release(); + body.close(); + } + } + + @Test + public void emptyStreamStopsImmediately() throws IOException { + Body body = new InputStreamBodyGenerator(new ByteArrayInputStream(new byte[0])).createBody(); + ByteBuf chunkBuffer = Unpooled.buffer(CHUNK_SIZE); + try { + assertEquals(BodyState.STOP, body.transferTo(chunkBuffer), "an empty stream must STOP immediately"); + assertEquals(0, chunkBuffer.readableBytes(), "nothing should be written for an empty stream"); + } finally { + chunkBuffer.release(); + body.close(); + } + } +} From 730f158ef981ce995d3a8da0a60e1aff23178c61 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Fri, 10 Jul 2026 11:53:51 +0200 Subject: [PATCH 2/2] Address review: drop the writableBytes() - 10 margin, fix docs and tests - transferTo now writes the full writable region instead of reserving a 10-byte margin. The margin was a legacy "safe guess", no consumer writes after transferTo, and it hid two bugs: a writable region < 10 threw an uncaught IllegalArgumentException and == 10 silently truncated the stream. Mirrors InputStreamMultipartPart, which writes the full writableBytes(). - Reword the comment: the no-staging/no-copy win applies to heap buffers; direct buffers still stage through a temp heap array internally. Reference InputStreamMultipartPart (FileLikeMultipartPart is abstract and never calls writeBytes(InputStream, int)). - Tests: use @RepeatedIfExceptionsTest like the sibling generator tests and add smallWritableRegionStillTransfers to lock the margin removal. --- .../generator/InputStreamBodyGenerator.java | 9 ++-- .../InputStreamBodyGeneratorTest.java | 41 +++++++++++++++---- 2 files changed, 39 insertions(+), 11 deletions(-) diff --git a/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java b/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java index eecbc75ce1..43b58e95ec 100644 --- a/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java +++ b/client/src/main/java/org/asynchttpclient/request/body/generator/InputStreamBodyGenerator.java @@ -71,12 +71,13 @@ public long getContentLength() { @Override public BodyState transferTo(ByteBuf target) { - // Read straight from the stream into the target buffer: no per-call staging byte[] and no extra - // copy (ByteBuf.writeBytes(InputStream, int) fills the buffer directly, like FileLikeMultipartPart). - // The "- 10" margin preserves the previous behaviour of never fully filling the writable region. + // Read straight from the stream into the target buffer instead of staging through a per-call byte[]. + // For heap target buffers this drops both the staging array and the copy; for direct buffers Netty + // still stages through a temporary heap array internally (InputStream can only read into a byte[]), + // so there the win is smaller. Mirrors InputStreamMultipartPart, which writes the full writable region. int read; try { - read = target.writeBytes(inputStream, target.writableBytes() - 10); + read = target.writeBytes(inputStream, target.writableBytes()); } catch (IOException ex) { LOGGER.warn("Unable to read", ex); return BodyState.STOP; diff --git a/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java b/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java index 6c9140840f..ed8b79f466 100644 --- a/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java +++ b/client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java @@ -15,11 +15,11 @@ */ package org.asynchttpclient.request.body.generator; +import io.github.artsok.RepeatedIfExceptionsTest; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.asynchttpclient.request.body.Body; import org.asynchttpclient.request.body.Body.BodyState; -import org.junit.jupiter.api.Test; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -31,14 +31,14 @@ /** * Covers {@link InputStreamBodyGenerator}'s {@link Body#transferTo(ByteBuf)}, which now reads straight from - * the stream into the target buffer (no per-call staging {@code byte[]} and no extra copy). The whole stream - * must still be transferred byte-for-byte, CONTINUE while data remains and STOP at EOF. + * the stream into the target buffer (dropping the per-call staging {@code byte[]} and copy for heap buffers). + * The whole stream must still be transferred byte-for-byte, CONTINUE while data remains and STOP at EOF. */ public class InputStreamBodyGeneratorTest { private static final int CHUNK_SIZE = 1024 * 8; - @Test + @RepeatedIfExceptionsTest(repeats = 5) public void streamsAllBytesAcrossMultipleReads() throws IOException { final byte[] src = new byte[3 * CHUNK_SIZE + 42]; new Random().nextBytes(src); @@ -62,9 +62,9 @@ public void streamsAllBytesAcrossMultipleReads() throws IOException { } } - @Test + @RepeatedIfExceptionsTest(repeats = 5) public void singleReadDrainsASmallStream() throws IOException { - final byte[] src = new byte[CHUNK_SIZE - 100]; // fits under writableBytes - 10, so one read drains it + final byte[] src = new byte[CHUNK_SIZE - 100]; // fits in one writable region, so one read drains it new Random().nextBytes(src); Body body = new InputStreamBodyGenerator(new ByteArrayInputStream(src)).createBody(); @@ -80,7 +80,7 @@ public void singleReadDrainsASmallStream() throws IOException { } } - @Test + @RepeatedIfExceptionsTest(repeats = 5) public void emptyStreamStopsImmediately() throws IOException { Body body = new InputStreamBodyGenerator(new ByteArrayInputStream(new byte[0])).createBody(); ByteBuf chunkBuffer = Unpooled.buffer(CHUNK_SIZE); @@ -92,4 +92,31 @@ public void emptyStreamStopsImmediately() throws IOException { body.close(); } } + + // Locks the removal of the old "writableBytes() - 10" margin: with a writable region of 10 or fewer bytes the + // margin made the transfer length 0 (or negative), so it silently STOPped without writing / threw. The stream + // must now still be drained through a tiny target buffer. + @RepeatedIfExceptionsTest(repeats = 5) + public void smallWritableRegionStillTransfers() throws IOException { + final byte[] src = new byte[25]; + new Random().nextBytes(src); + + Body body = new InputStreamBodyGenerator(new ByteArrayInputStream(src)).createBody(); + ByteBuf chunkBuffer = Unpooled.buffer(10, 10); // writableBytes() == 10, the old margin's boundary + ByteArrayOutputStream collected = new ByteArrayOutputStream(); + try { + BodyState state; + while ((state = body.transferTo(chunkBuffer)) != BodyState.STOP) { + assertEquals(BodyState.CONTINUE, state, "a stream with data left must report CONTINUE"); + byte[] b = new byte[chunkBuffer.readableBytes()]; + chunkBuffer.readBytes(b); + collected.write(b); + chunkBuffer.clear(); + } + assertArrayEquals(src, collected.toByteArray(), "the whole stream must drain through a tiny buffer"); + } finally { + chunkBuffer.release(); + body.close(); + } + } }