- * The stream will not be closed by the SDK. It is up to to caller of this method to close the stream. The stream - * should not be read outside of the SDK (by another thread) as it will change the state of the {@link InputStream} and + * The stream will not be closed by the SDK. It is up to caller of this method to close the stream. The stream + * should not be read outside the SDK (by another thread) as it will change the state of the {@link InputStream} and * could tamper with the sending of the request. *
* To support resetting via {@link ContentStreamProvider}, this uses {@link InputStream#reset()} and uses a read limit of * 128 KiB. If you need more control, use {@link #fromContentProvider(ContentStreamProvider, long, String)} or * {@link #fromContentProvider(ContentStreamProvider, String)}. * + *
+ * It is recommended to provide a stream that supports mark and reset for retry. If the stream does not support mark and + * reset, an {@link IllegalStateException} will be thrown during retry. + * * @param inputStream Input stream to send to the service. The stream will not be closed by the SDK. * @param contentLength Content length of data in input stream. If a content length smaller than the actual size of the * object is set, the client will truncate the stream to the specified content length and only send @@ -135,23 +138,10 @@ public static RequestBody fromFile(File file) { * @return RequestBody instance. */ public static RequestBody fromInputStream(InputStream inputStream, long contentLength) { - IoUtils.markStreamWithMaxReadLimit(inputStream); - InputStream nonCloseable = nonCloseableInputStream(inputStream); - ContentStreamProvider provider = new ContentStreamProvider() { - @Override - public InputStream newStream() { - if (nonCloseable.markSupported()) { - invokeSafely(nonCloseable::reset); - } - return nonCloseable; - } - - @Override - public String name() { - return ProviderType.STREAM.getName(); - } - }; - return fromContentProvider(provider, contentLength, Mimetype.MIMETYPE_OCTET_STREAM); + ContentStreamProvider contentStreamProvider = ContentStreamProvider.fromInputStream( + nonCloseableInputStream(inputStream)); + return fromContentProvider(contentStreamProvider, + contentLength, Mimetype.MIMETYPE_OCTET_STREAM); } /** diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/sync/RequestBodyTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/sync/RequestBodyTest.java index 67157c8bdcee..0ae1726378de 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/sync/RequestBodyTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/sync/RequestBodyTest.java @@ -39,6 +39,7 @@ import software.amazon.awssdk.checksums.SdkChecksum; import software.amazon.awssdk.core.internal.sync.BufferingContentStreamProvider; import software.amazon.awssdk.core.internal.util.Mimetype; +import software.amazon.awssdk.testutils.RandomInputStream; import software.amazon.awssdk.utils.BinaryUtils; import software.amazon.awssdk.utils.IoUtils; import software.amazon.awssdk.utils.StringInputStream; @@ -163,6 +164,17 @@ public void fromInputStream_streamSupportsReset_resetsTheStream() { assertThat(getCrc32(requestBody.contentStreamProvider().newStream())).isEqualTo(streamCrc32); } + @Test + public void fromInputStream_streamNotSupportReset_shouldThrowException() { + RandomInputStream stream = new RandomInputStream(100); + assertThat(stream.markSupported()).isFalse(); + RequestBody requestBody = RequestBody.fromInputStream(stream, 100); + IoUtils.drainInputStream(requestBody.contentStreamProvider().newStream()); + assertThatThrownBy(() -> requestBody.contentStreamProvider().newStream()) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Content input stream does not support mark/reset"); + } + private static String getCrc32(InputStream inputStream) { byte[] buff = new byte[1024]; int read;