Allows users to configure contentType for BlockingInputStreamAsyncRequestBody (#5234)

* Allows users to configure contentType for BlockingInputStreamAsyncRequestBody (#5143)

* Run new-change (#5143)

* Add tests for BlockingInputStreamAsyncRequestBody (#5143)

* Update tests

---------

Co-authored-by: Kirill Chaykin <[email protected]>
Co-authored-by: hdavidh <[email protected]>
Co-authored-by: David Ho <[email protected]>

Author: 3 people

.changes/next-release/feature-AWSS3-bb458a5.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "Amazon S3",
+    "contributor": "chaykin",
+    "description": "Allow user to configure content type for BlockingInputStreamAsyncRequestBody"
+}

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

@@ -384,8 +384,8 @@ static AsyncRequestBody fromInputStream(Consumer<AsyncRequestBodyFromInputStream
     /**
      * Creates a {@link BlockingInputStreamAsyncRequestBody} to use for writing an input stream to the downstream service.
      *
-     * <p>By default, it will time out if streaming hasn't started within 10 seconds, and you can configure the timeout
-     * via {@link BlockingInputStreamAsyncRequestBody#builder()}
+     * <p>By default, it will time out if streaming hasn't started within 10 seconds, and use application/octet-stream as
+     * content type. You can configure it via {@link BlockingInputStreamAsyncRequestBody#builder()}
      * <p><b>Example Usage</b>
      *
      * <p>

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

@@ -26,6 +26,7 @@
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.core.exception.NonRetryableException;
 import software.amazon.awssdk.core.internal.io.SdkLengthAwareInputStream;
+import software.amazon.awssdk.core.internal.util.Mimetype;
 import software.amazon.awssdk.core.internal.util.NoopSubscription;
 import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.async.InputStreamConsumingPublisher;
@@ -39,14 +40,17 @@
 @SdkPublicApi
 public final class BlockingInputStreamAsyncRequestBody implements AsyncRequestBody {
     private static final Duration DEFAULT_SUBSCRIBE_TIMEOUT = Duration.ofSeconds(10);
+    private static final String DEFAULT_CONTENT_TYPE = Mimetype.MIMETYPE_OCTET_STREAM;
     private final InputStreamConsumingPublisher delegate = new InputStreamConsumingPublisher();
     private final CountDownLatch subscribedLatch = new CountDownLatch(1);
     private final AtomicBoolean subscribeCalled = new AtomicBoolean(false);
     private final Long contentLength;
+    private final String contentType;
     private final Duration subscribeTimeout;
 
     BlockingInputStreamAsyncRequestBody(Builder builder) {
         this.contentLength = builder.contentLength;
+        this.contentType = builder.contentType != null ? builder.contentType : DEFAULT_CONTENT_TYPE;
         this.subscribeTimeout = Validate.isPositiveOrNull(builder.subscribeTimeout, "subscribeTimeout") != null ?
                                 builder.subscribeTimeout :
                                 DEFAULT_SUBSCRIBE_TIMEOUT;
@@ -64,6 +68,11 @@ public Optional<Long> contentLength() {
         return Optional.ofNullable(contentLength);
     }
 
+    @Override
+    public String contentType() {
+        return contentType;
+    }
+
     /**
      * Block the calling thread and write the provided input stream to the downstream service.
      *
@@ -123,6 +132,7 @@ private void waitForSubscriptionIfNeeded() throws InterruptedException {
     public static final class Builder {
         private Duration subscribeTimeout;
         private Long contentLength;
+        private String contentType;
 
         private Builder() {
         }
@@ -152,6 +162,17 @@ public Builder contentLength(Long contentLength) {
             return this;
         }
 
+        /**
+         * The content type of the output stream.
+         *
+         * @param contentType the content type
+         * @return Returns a reference to this object so that method calls can be chained together.
+         */
+        public Builder contentType(String contentType) {
+            this.contentType = contentType;
+            return this;
+        }
+
         public BlockingInputStreamAsyncRequestBody build() {
             return new BlockingInputStreamAsyncRequestBody(this);
         }

core/sdk-core/src/test/java/software/amazon/awssdk/core/async/BlockingInputStreamAsyncRequestBodyTest.java

@@ -23,6 +23,7 @@
 import java.io.ByteArrayInputStream;
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import org.junit.jupiter.api.Test;
@@ -76,4 +77,26 @@ public void doBlockingWrite_writesToSubscriber() {
         }
     }
 
+    @Test
+    public void builder_withContentType_buildsCorrectly() {
+        BlockingInputStreamAsyncRequestBody.Builder requestBodyBuilder = BlockingInputStreamAsyncRequestBody.builder();
+        BlockingInputStreamAsyncRequestBody requestBody = requestBodyBuilder
+            .contentLength(20L)
+            .contentType("text/plain")
+            .build();
+
+        assertThat(requestBody.contentLength()).isEqualTo(Optional.of(20L));
+        assertThat(requestBody.contentType()).isEqualTo("text/plain");
+    }
+
+    @Test
+    public void builder_withoutContentType_defaultsToOctetStream() {
+        BlockingInputStreamAsyncRequestBody.Builder requestBodyBuilder = BlockingInputStreamAsyncRequestBody.builder();
+        BlockingInputStreamAsyncRequestBody requestBody = requestBodyBuilder
+            .contentLength(20L)
+            .build();
+
+        assertThat(requestBody.contentLength()).isEqualTo(Optional.of(20L));
+        assertThat(requestBody.contentType()).isEqualTo("application/octet-stream");
+    }
 }

services/s3/src/it/java/software/amazon/awssdk/services/s3/PutObjectIntegrationTest.java

@@ -28,19 +28,27 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
 import software.amazon.awssdk.core.sync.RequestBody;
 import software.amazon.awssdk.http.ContentStreamProvider;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 
 /**
  * Integration tests for {@code PutObject}.
  */
 public class PutObjectIntegrationTest extends S3IntegrationTestBase {
     private static final String BUCKET = temporaryBucketName(PutObjectIntegrationTest.class);
-    private static final String KEY = "key";
+    private static final String ASYNC_KEY = "async-key";
+    private static final String SYNC_KEY = "sync-key";
+
     private static final byte[] CONTENT = "Hello".getBytes(StandardCharsets.UTF_8);
+    private static final String TEXT_CONTENT_TYPE = "text/plain";
 
     @BeforeClass
     public static void setUp() throws Exception {
@@ -56,13 +64,36 @@ public static void tearDown() {
     @Test
     public void objectInputStreamsAreClosed() {
         TestContentProvider provider = new TestContentProvider(CONTENT);
-        s3.putObject(r -> r.bucket(BUCKET).key(KEY), RequestBody.fromContentProvider(provider, CONTENT.length, "binary/octet-stream"));
+        s3.putObject(r -> r.bucket(BUCKET).key(SYNC_KEY),
+                     RequestBody.fromContentProvider(provider, CONTENT.length, "binary/octet-stream"));
 
         for (CloseTrackingInputStream is : provider.getCreatedStreams()) {
             assertThat(is.isClosed()).isTrue();
         }
     }
 
+    @Test
+    public void blockingInputStreamAsyncRequestBody_withContentType_isHonored() {
+        BlockingInputStreamAsyncRequestBody requestBody =
+            BlockingInputStreamAsyncRequestBody.builder()
+                                               .contentLength((long) CONTENT.length)
+                                               .contentType(TEXT_CONTENT_TYPE)
+                                               .build();
+
+        PutObjectRequest.Builder request = PutObjectRequest.builder()
+                                                           .bucket(BUCKET)
+                                                           .key(ASYNC_KEY);
+
+        CompletableFuture<PutObjectResponse> responseFuture = s3Async.putObject(request.build(), requestBody);
+        requestBody.writeInputStream(new ByteArrayInputStream(CONTENT));
+        responseFuture.join();
+
+        HeadObjectResponse response = s3Async.headObject(r -> r.bucket(BUCKET).key(ASYNC_KEY)).join();
+
+        assertThat(response.contentLength()).isEqualTo(CONTENT.length);
+        assertThat(response.contentType()).isEqualTo(TEXT_CONTENT_TYPE);
+    }
+
     @Test
     public void s3Client_usingHttpAndDisableChunkedEncoding() {
         try (S3Client s3Client = s3ClientBuilder()
@@ -71,7 +102,7 @@ public void s3Client_usingHttpAndDisableChunkedEncoding() {
                                                  .chunkedEncodingEnabled(false)
                                                  .build())
             .build()) {
-            assertThat(s3Client.putObject(b -> b.bucket(BUCKET).key(KEY), RequestBody.fromBytes(
+            assertThat(s3Client.putObject(b -> b.bucket(BUCKET).key(SYNC_KEY), RequestBody.fromBytes(
                 "helloworld".getBytes()))).isNotNull();
         }
     }