Use maximumBufferSize in ByteArraySplittingTransformer

Author: davidh44

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ByteArrayAsyncResponseTransformer.java

@@ -28,6 +28,8 @@
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.utils.BinaryUtils;
+import software.amazon.awssdk.utils.Validate;
+import software.amazon.awssdk.utils.async.DelegatingBufferingSubscriber;
 
 /**
  * Implementation of {@link AsyncResponseTransformer} that dumps content into a byte array and supports further
@@ -61,16 +63,24 @@ public void onStream(SdkPublisher<ByteBuffer> publisher) {
         publisher.subscribe(new BaosSubscriber(cf));
     }
 
+    public void onStream(SdkPublisher<ByteBuffer> publisher, long maximumBufferSize) {
+        publisher.subscribe(DelegatingBufferingSubscriber.builder()
+                                                         .maximumBufferInBytes(maximumBufferSize)
+                                                         .delegate(new BaosSubscriber(cf))
+                                                         .build());
+    }
+
     @Override
     public void exceptionOccurred(Throwable throwable) {
         cf.completeExceptionally(throwable);
     }
 
     @Override
     public SplitResult<ResponseT, ResponseBytes<ResponseT>> split(SplittingTransformerConfiguration splitConfig) {
+        Validate.notNull(splitConfig, "splitConfig must not be null");
         CompletableFuture<ResponseBytes<ResponseT>> future = new CompletableFuture<>();
         SdkPublisher<AsyncResponseTransformer<ResponseT, ResponseT>> transformer =
-            new ByteArraySplittingTransformer<>(this, future);
+            new ByteArraySplittingTransformer<>(this, future, splitConfig.bufferSizeInBytes());
         return AsyncResponseTransformer.SplitResult.<ResponseT, ResponseBytes<ResponseT>>builder()
                                                    .publisher(transformer)
                                                    .resultFuture(future)

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ByteArraySplittingTransformer.java

@@ -34,6 +34,7 @@
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.utils.CompletableFutureUtils;
 import software.amazon.awssdk.utils.Logger;
+import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.async.SimplePublisher;
 
 @SdkInternalApi
@@ -73,9 +74,18 @@ public class ByteArraySplittingTransformer<ResponseT> implements SdkPublisher<As
 
     private final Map<Integer, ByteBuffer> buffers;
 
+    /**
+     * The buffer size used to buffer the content received from the downstream subscriber
+     */
+    private final long maximumBufferInBytes;
+
     public ByteArraySplittingTransformer(AsyncResponseTransformer<ResponseT, ResponseBytes<ResponseT>>
                                              upstreamResponseTransformer,
-                                         CompletableFuture<ResponseBytes<ResponseT>> resultFuture) {
+                                         CompletableFuture<ResponseBytes<ResponseT>> resultFuture,
+                                         Long maximumBufferSizeInBytes) {
+        Validate.notNull(maximumBufferSizeInBytes, "maximumBufferSizeInBytes");
+        this.maximumBufferInBytes = Validate.isPositive(
+            maximumBufferSizeInBytes, "maximumBufferSizeInBytes");
         this.upstreamResponseTransformer = upstreamResponseTransformer;
         this.resultFuture = resultFuture;
         this.buffers = new ConcurrentHashMap<>();
@@ -223,7 +233,7 @@ public void onResponse(ResponseT response) {
 
         @Override
         public void onStream(SdkPublisher<ByteBuffer> publisher) {
-            delegate.onStream(publisher);
+            delegate.onStream(publisher, maximumBufferInBytes);
             synchronized (lock) {
                 if (!onStreamCalled) {
                     onStreamCalled = true;

services/s3/src/main/java/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.java

@@ -90,10 +90,6 @@ public Long minimumPartSizeInBytes() {
 
     /**
      * The maximum memory, in bytes, that the SDK will use to buffer requests content into memory.
-     * <p>
-     * This setting is not supported and will be ignored when downloading to a byte array, i.e., when providing a
-     * {@link ByteArrayAsyncResponseTransformer}.
-     *
      * @return the value of the configured maximum memory usage.
      */
     public Long apiCallBufferSizeInBytes() {
@@ -163,9 +159,6 @@ public interface Builder extends CopyableBuilder<Builder, MultipartConfiguration
          * Increasing this value may lead to better performance at the cost of using more memory.
          * <p>
          * Default value: If not specified, the SDK will use the equivalent of four parts worth of memory, so 32 Mib by default.
-         * <p>
-         * This setting is not supported and will be ignored when downloading to a byte array, i.e., when providing a
-         * {@link ByteArrayAsyncResponseTransformer}.
          *
          * @param apiCallBufferSizeInBytes the value of the maximum memory usage.
          * @return an instance of this builder.