When the length is known up front this allocates the result array once and reads straight into it. That avoids + * the staging buffer, the doubling reallocations and the final defensive copy that + * {@link IoUtils#toByteArray(InputStream)} needs in order to handle an unknown length. + * + *
The stream is deliberately not closed, matching the previous behavior: some + * {@link ContentStreamProvider} implementations (notably {@link ContentStreamProvider#fromInputStream(InputStream)}) + * hand back the same stream on every call and rely on mark/reset, so closing it here would break the next retry + * attempt. + * + * @param presizeHint expected length, or a negative number if unknown. + */ + private static byte[] readFully(InputStream stream, int presizeHint) throws IOException { + if (presizeHint < 0) { + return IoUtils.toByteArray(stream); + } + + byte[] buffer = new byte[presizeHint]; + int read = readUpTo(stream, buffer); + + if (read < presizeHint) { + // Stream was shorter than advertised. Trim rather than pad the body with trailing zeros. + byte[] trimmed = new byte[read]; + System.arraycopy(buffer, 0, trimmed, 0, read); + return trimmed; + } + + // Either the stream is longer than advertised, or it is longer than MAX_PRESIZED_ALLOCATION. Read the remainder + // instead of silently truncating the body. + int next = stream.read(); + if (next < 0) { + return buffer; + } + return readRemainder(stream, buffer, next); + } + + /** + * Fills {@code buffer} as far as the stream allows, returning the number of bytes read. + */ + private static int readUpTo(InputStream stream, byte[] buffer) throws IOException { + int read = 0; + while (read < buffer.length) { + int n = stream.read(buffer, read, buffer.length - read); + if (n < 0) { + break; + } + read += n; + } + return read; + } + + private static byte[] readRemainder(InputStream stream, byte[] alreadyRead, int nextByte) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(alreadyRead.length * 2); + out.write(alreadyRead, 0, alreadyRead.length); + out.write(nextByte); + + byte[] chunk = new byte[GROWING_READ_BUFFER_SIZE]; + int n; + while ((n = stream.read(chunk)) != -1) { + out.write(chunk, 0, n); + } + return out.toByteArray(); + } + + /** + * The marshallers and the body-rewriting pipeline stages set {@code Content-Length} for in-memory bodies, so this is + * normally present and exact. + * + * @return the length to pre-allocate, or a negative number if it could not be determined. + */ + private static int presizeHint(SdkHttpFullRequest request, int maxPresizedAllocation) { + return request.firstMatchingHeader(Header.CONTENT_LENGTH) + .map(contentLength -> parsePresizeHint(contentLength, maxPresizedAllocation)) + .orElse(-1); + } + + private static int parsePresizeHint(String contentLength, int maxPresizedAllocation) { + long parsed; + try { + parsed = Long.parseLong(contentLength); + } catch (NumberFormatException e) { + return -1; + } + + if (parsed < 0) { + return -1; + } + return (int) Math.min(parsed, maxPresizedAllocation); + } + private class SubscriptionImpl implements Subscription { private boolean running = true; private final Subscriber s; diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/async/SimpleHttpContentPublisherTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/async/SimpleHttpContentPublisherTest.java new file mode 100644 index 000000000000..1ca4de38845f --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/async/SimpleHttpContentPublisherTest.java @@ -0,0 +1,421 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. 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. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.core.internal.http.async; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import software.amazon.awssdk.http.ContentStreamProvider; +import software.amazon.awssdk.http.Header; +import software.amazon.awssdk.http.SdkHttpFullRequest; +import software.amazon.awssdk.http.SdkHttpMethod; + +class SimpleHttpContentPublisherTest { + + private static final byte[] BODY = "The quick brown fox jumps over the lazy dog".getBytes(StandardCharsets.UTF_8); + + @Test + void contentLength_whenNoContentStreamProvider_isZero() { + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestBuilder().build()); + + assertThat(publisher.contentLength()).hasValue(0L); + } + + @Test + void subscribe_whenNoContentStreamProvider_publishesEmptyBuffer() { + CollectingSubscriber subscriber = drain(new SimpleHttpContentPublisher(requestBuilder().build())); + + assertThat(subscriber.content()).isEmpty(); + assertThat(subscriber.bufferCount()).isEqualTo(1); + } + + @Test + void subscribe_whenContentLengthMatchesBody_publishesFullBody() { + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length)); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + @Test + void subscribe_whenContentLengthHeaderAbsent_publishesFullBody() { + SdkHttpFullRequest request = requestBuilder().contentStreamProvider(() -> new ByteArrayInputStream(BODY)).build(); + + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(request); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * An unusable Content-Length must fall back to reading the stream rather than failing or truncating. + */ + @ParameterizedTest + @ValueSource(strings = {"", " ", "not-a-number", "-1", "12.5", "99999999999999999999999999"}) + void subscribe_whenContentLengthHeaderIsUnusable_publishesFullBody(String contentLength) { + SdkHttpFullRequest request = requestBuilder().contentStreamProvider(() -> new ByteArrayInputStream(BODY)) + .putHeader(Header.CONTENT_LENGTH, contentLength) + .build(); + + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(request); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + @Test + void subscribe_whenContentLengthIsZeroAndBodyEmpty_publishesEmptyBuffer() { + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(new byte[0], 0)); + + assertThat(publisher.contentLength()).hasValue(0L); + assertThat(drain(publisher).content()).isEmpty(); + } + + /** + * The pre-sized read has to loop, because an {@link InputStream} may return fewer bytes than asked for. + */ + @Test + void subscribe_whenStreamReturnsPartialReads_publishesFullBody() { + SdkHttpFullRequest request = + requestBuilder().contentStreamProvider(() -> new OneByteAtATimeInputStream(BODY)) + .putHeader(Header.CONTENT_LENGTH, Integer.toString(BODY.length)) + .build(); + + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(request); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * Content-Length larger than the stream: publish what is actually there rather than padding with trailing zeros. + */ + @Test + void subscribe_whenContentLengthLongerThanStream_publishesOnlyActualBytes() { + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length + 100)); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * Content-Length smaller than the stream: read the remainder rather than truncating the body. + */ + @Test + void subscribe_whenContentLengthShorterThanStream_publishesFullBody() { + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, 5)); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * A body larger than the pre-size ceiling still has to be published in full. + */ + @Test + void subscribe_whenBodyLargerThanPresizeCeiling_publishesFullBody() { + SimpleHttpContentPublisher publisher = + new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length), 8); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * The ceiling exists so that a bogus Content-Length cannot drive a huge allocation. The body still has to be + * published in full, and the reported length has to come from the body rather than the header. + */ + @Test + void subscribe_whenContentLengthFarExceedsPresizeCeiling_publishesFullBody() { + SimpleHttpContentPublisher publisher = + new SimpleHttpContentPublisher(requestWithBody(BODY, Integer.MAX_VALUE), 8); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * Same as above but through the production ceiling, so a {@code Content-Length} of ~2 GiB must not be allocated. + */ + @Test + void subscribe_whenContentLengthIsIntMaxValue_publishesFullBodyWithoutExhaustingMemory() { + SimpleHttpContentPublisher publisher = + new SimpleHttpContentPublisher(requestWithBody(BODY, Integer.MAX_VALUE)); + + assertThat(publisher.contentLength()).hasValue((long) BODY.length); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + @Test + void subscribe_whenSubscribedTwice_eachSubscriberReceivesFullBody() { + SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length)); + + assertThat(drain(publisher).content()).isEqualTo(BODY); + assertThat(drain(publisher).content()).isEqualTo(BODY); + } + + /** + * A retry constructs a new publisher from the same request. Both attempts must see identical bytes. + * + *
Uses a stream that fails reads after {@code close()} and is exposed through
+ * {@link ContentStreamProvider#fromInputStream(InputStream)}, which hands back the same stream on every
+ * call and relies on mark/reset. Closing the stream while buffering would break the second attempt.
+ */
+ @Test
+ void construct_whenProviderReusesStreamAcrossAttempts_bothAttemptsSeeSameBody() {
+ CloseAwareInputStream stream = new CloseAwareInputStream(BODY);
+ SdkHttpFullRequest request = requestBuilder().contentStreamProvider(ContentStreamProvider.fromInputStream(stream))
+ .putHeader(Header.CONTENT_LENGTH, Integer.toString(BODY.length))
+ .build();
+
+ byte[] firstAttempt = drain(new SimpleHttpContentPublisher(request)).content();
+ byte[] secondAttempt = drain(new SimpleHttpContentPublisher(request)).content();
+
+ assertThat(firstAttempt).isEqualTo(BODY);
+ assertThat(secondAttempt).isEqualTo(BODY);
+ assertThat(stream.closeCount()).isZero();
+ }
+
+ @Test
+ void construct_whenContentLengthHeaderAbsent_doesNotCloseProviderStream() {
+ CloseAwareInputStream stream = new CloseAwareInputStream(BODY);
+ SdkHttpFullRequest request = requestBuilder().contentStreamProvider(() -> stream).build();
+
+ assertThat(drain(new SimpleHttpContentPublisher(request)).content()).isEqualTo(BODY);
+ assertThat(stream.closeCount()).isZero();
+ }
+
+ @Test
+ void request_whenDemandIsNotPositive_signalsError() {
+ SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length));
+ CollectingSubscriber subscriber = new CollectingSubscriber(0);
+ publisher.subscribe(subscriber);
+
+ subscriber.subscription().request(0);
+
+ assertThat(subscriber.error()).isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Demand must be positive");
+ assertThat(subscriber.bufferCount()).isZero();
+ assertThat(subscriber.completed()).isFalse();
+ }
+
+ @Test
+ void request_whenCalledRepeatedly_publishesBodyOnce() {
+ SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length));
+ CollectingSubscriber subscriber = new CollectingSubscriber(0);
+ publisher.subscribe(subscriber);
+
+ subscriber.subscription().request(1);
+ subscriber.subscription().request(1);
+ subscriber.subscription().request(Long.MAX_VALUE);
+
+ assertThat(subscriber.bufferCount()).isEqualTo(1);
+ assertThat(subscriber.content()).isEqualTo(BODY);
+ assertThat(subscriber.completeCount()).isEqualTo(1);
+ }
+
+ @Test
+ void cancel_whenCancelledBeforeRequest_publishesNothing() {
+ SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length));
+ CollectingSubscriber subscriber = new CollectingSubscriber(0);
+ publisher.subscribe(subscriber);
+
+ subscriber.subscription().cancel();
+ subscriber.subscription().request(Long.MAX_VALUE);
+
+ assertThat(subscriber.bufferCount()).isZero();
+ assertThat(subscriber.completed()).isFalse();
+ assertThat(subscriber.error()).isNull();
+ }
+
+ /**
+ * The published buffer has to line up with what {@link SimpleHttpContentPublisher#contentLength()} advertises, since
+ * the HTTP clients use the two together to frame the request.
+ */
+ @Test
+ void subscribe_whenBodyPublished_bufferRemainingMatchesContentLength() {
+ SimpleHttpContentPublisher publisher = new SimpleHttpContentPublisher(requestWithBody(BODY, BODY.length));
+
+ CollectingSubscriber subscriber = drain(publisher);
+
+ assertThat(subscriber.totalRemaining()).isEqualTo(publisher.contentLength().get());
+ }
+
+ private static SdkHttpFullRequest.Builder requestBuilder() {
+ return SdkHttpFullRequest.builder()
+ .uri(URI.create("https://aws.amazon.com"))
+ .method(SdkHttpMethod.POST);
+ }
+
+ private static SdkHttpFullRequest requestWithBody(byte[] body, long advertisedContentLength) {
+ return requestBuilder().contentStreamProvider(() -> new ByteArrayInputStream(body))
+ .putHeader(Header.CONTENT_LENGTH, Long.toString(advertisedContentLength))
+ .build();
+ }
+
+ private static CollectingSubscriber drain(SimpleHttpContentPublisher publisher) {
+ CollectingSubscriber subscriber = new CollectingSubscriber(Long.MAX_VALUE);
+ publisher.subscribe(subscriber);
+ assertThat(subscriber.error()).isNull();
+ assertThat(subscriber.completed()).isTrue();
+ return subscriber;
+ }
+
+ private static final class CollectingSubscriber implements Subscriber