Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
@@ -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;

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: the existing tests in this package (ByteArrayBodyGeneratorTest and FeedableBodyGeneratorTest) use @RepeatedIfExceptionsTest(repeats = 5) on every test method, so using plain @Test breaks the local convention.

ByteArrayBodyGeneratorTest also already has the same single-read and multi-read scaffold (same chunk size, same 3 * CHUNK_SIZE + 42 sizing, and the same drain loop), while FeedableBodyGeneratorTest provides a readFromBody helper that covers the short-stream cases. It would be worth aligning with the existing annotation and reusing the existing test patterns rather than introducing a second drain idiom to the package.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

@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();
}
}
}
Loading