Read InputStream bodies straight into the buffer, no staging array (#2233)

Motivation:

`InputStreamBodyGenerator.Body.transferTo` allocated a new `byte[]` on
every invocation, sized to the target `ByteBuf`'s writable region, then
read the `InputStream` into that array before copying the bytes into the
`ByteBuf`. This incurred both a per-chunk allocation and an unnecessary
copy. Additionally, the implementation reserved a 10-byte margin
(`writableBytes() - 10`), which could cause `NegativeArraySizeException`
or prematurely stop transfers when the target had 10 or fewer writable
bytes.

Modification:

`transferTo` now reads directly from the `InputStream` into the target
`ByteBuf` using `ByteBuf.writeBytes(InputStream, int)`, eliminating the
intermediate staging array, the extra copy, and the per-instance chunk
field.

The state machine is unchanged: each call performs a single stream read,
returns `CONTINUE` while data remains, and returns `STOP` at EOF or on
an I/O error (with logging unchanged). The legacy 10-byte writable
margin has been removed, allowing each transfer to use the full writable
capacity and correctly handle small writable regions.

Although AsyncHttpClient's internal request path uses
`NettyInputStreamBody`, this implementation remains part of the public
`InputStreamBodyGenerator.createBody()` API and is used by external
callers.

Result:

Reduces allocation and copy overhead during transfers, correctly
supports small writable buffers, preserves existing transfer semantics,
and adds tests covering multi-read transfers, single-read completion,
empty streams, and transfers through a 10-byte buffer. No public API
changes.

Author: pavel-ptashyts

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,18 @@ 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 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 = inputStream.read(chunk);
+                read = target.writeBytes(inputStream, target.writableBytes());
             } 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

client/src/test/java/org/asynchttpclient/request/body/generator/InputStreamBodyGeneratorTest.java

@@ -0,0 +1,122 @@
+/*
+ *    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.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 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 (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;
+
+    @RepeatedIfExceptionsTest(repeats = 5)
+    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();
+        }
+    }
+
+    @RepeatedIfExceptionsTest(repeats = 5)
+    public void singleReadDrainsASmallStream() throws IOException {
+        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();
+        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();
+        }
+    }
+
+    @RepeatedIfExceptionsTest(repeats = 5)
+    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();
+        }
+    }
+
+    // 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();
+        }
+    }
+}