PR comments

- renamed EmittingSubscription, mark it ThreadSafe
- Added comments
- some other renaming

Author: L-Applin

…sync/ThreadSafeEmittingSubscription.java …internal/async/EmittingSubscription.javacore/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java renamed to core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ThreadSafeEmittingSubscription.java renamed to core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/EmittingSubscription.java

@@ -21,11 +21,18 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.annotations.ThreadSafe;
 import software.amazon.awssdk.utils.Logger;
 
+/**
+ * Subscription which can emit {@link Subscriber#onNext(T)} signals to a subscriber, based on the demand received with the
+ * {@link Subscription#request(long)}. It tracks the outstandingDemand that has not yet been fulfilled and used a Supplier
+ * passed to it to create the object it needs to emit.
+ * @param <T> the type of obejct to emit to the subscriber.
+ */
 @SdkInternalApi
-public class ThreadSafeEmittingSubscription<T> implements Subscription {
+@ThreadSafe
+public final class EmittingSubscription<T> implements Subscription {
 
     private Subscriber<? super T> downstreamSubscriber;
     private final AtomicBoolean emitting = new AtomicBoolean(false);
@@ -36,7 +43,7 @@ public class ThreadSafeEmittingSubscription<T> implements Subscription {
     private final Logger log;
 
 
-    private ThreadSafeEmittingSubscription(Builder<T> builder) {
+    private EmittingSubscription(Builder<T> builder) {
         this.downstreamSubscriber = builder.downstreamSubscriber;
         this.outstandingDemand = builder.outstandingDemand;
         this.onCancel = builder.onCancel;
@@ -113,7 +120,7 @@ public static class Builder<T> {
         private Subscriber<? super T> downstreamSubscriber;
         private AtomicLong outstandingDemand = new AtomicLong(0);
         private AtomicBoolean isCancelled = new AtomicBoolean(false);
-        private Logger log = Logger.loggerFor(ThreadSafeEmittingSubscription.class);
+        private Logger log = Logger.loggerFor(EmittingSubscription.class);
         private Runnable onCancel;
         private Supplier<T> supplier;
 
@@ -147,8 +154,8 @@ public Builder<T> supplier(Supplier<T> supplier) {
             return this;
         }
 
-        public ThreadSafeEmittingSubscription<T> build() {
-            return new ThreadSafeEmittingSubscription<>(this);
+        public EmittingSubscription<T> build() {
+            return new EmittingSubscription<>(this);
         }
     }
 

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

@@ -60,16 +60,14 @@ public FileAsyncResponseTransformerPublisher(FileAsyncResponseTransformer<?> res
 
     @Override
     public void subscribe(Subscriber<? super AsyncResponseTransformer<T, T>> s) {
-        if (s == null) {
-            throw new NullPointerException("Subscription must not be null");
-        }
+        Validate.notNull(s, "Subscriber must not be null");
         this.subscriber = s;
-        s.onSubscribe(ThreadSafeEmittingSubscription.<AsyncResponseTransformer<T, T>>builder()
-                                                    .downstreamSubscriber(s)
-                                                    .onCancel(this::onCancel)
-                                                    .log(log)
-                                                    .supplier(this::createTransformer)
-                                                    .build());
+        s.onSubscribe(EmittingSubscription.<AsyncResponseTransformer<T, T>>builder()
+                                          .downstreamSubscriber(s)
+                                          .onCancel(this::onCancel)
+                                          .log(log)
+                                          .supplier(this::createTransformer)
+                                          .build());
     }
 
     private AsyncResponseTransformer<T, T> createTransformer() {
@@ -110,8 +108,7 @@ public void onResponse(T response) {
             if (!contentRangeList.isPresent()) {
                 if (subscriber != null) {
                     IllegalStateException e = new IllegalStateException("Content range header is missing");
-                    subscriber.onError(e);
-                    future.completeExceptionally(e);
+                    handleError(e);
                 }
                 return;
             }
@@ -121,8 +118,7 @@ public void onResponse(T response) {
             if (!contentRangePair.isPresent()) {
                 if (subscriber != null) {
                     IllegalStateException e = new IllegalStateException("Could not parse content range header " + contentRange);
-                    subscriber.onError(e);
-                    future.completeExceptionally(e);
+                    handleError(e);
                 }
                 return;
             }
@@ -135,6 +131,11 @@ public void onResponse(T response) {
             delegate.onResponse(response);
         }
 
+        private void handleError(Throwable e) {
+            subscriber.onError(e);
+            future.completeExceptionally(e);
+        }
+
         private AsyncResponseTransformer<T, T> getDelegateTransformer(Long startAt) {
             if (transformerCount.get() == 0) {
                 return AsyncResponseTransformer.toFile(path, initialConfig);
@@ -154,7 +155,7 @@ private AsyncResponseTransformer<T, T> getDelegateTransformer(Long startAt) {
                         .position(initialOffset + startAt));
                     return AsyncResponseTransformer.toFile(path, newConfig);
                 }
-                // APPEND mode is not supported for non-serial operations,
+                // As per design specification, APPEND mode is not supported for non-serial operations
                 case CREATE_OR_APPEND_TO_EXISTING:
                 default:
                     throw new UnsupportedOperationException("Unsupported fileWriteOption: " + initialConfig.fileWriteOption());

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/MultipartConfigurationResolver.java

@@ -27,6 +27,8 @@
 public final class MultipartConfigurationResolver {
 
     private static final long DEFAULT_MIN_PART_SIZE = 8L * 1024 * 1024;
+
+    // Using 50 as the default since maxConcurrency for http client is also 50
     private static final int DEFAULT_MAX_IN_FLIGHT_PARTS = 50;
 
     private final long minimalPartSizeInBytes;

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/NonLinearMultipartDownloaderSubscriber.java

@@ -122,7 +122,7 @@ public class NonLinearMultipartDownloaderSubscriber
     /**
      * Tracks if the first request was sent
      */
-    private final AtomicBoolean firstRequestSent = new AtomicBoolean(false);
+    private final AtomicBoolean isFirstRequestSent = new AtomicBoolean(false);
 
     /**
      * Tracks the total number of transformers requested from the subscription
@@ -139,7 +139,7 @@ public class NonLinearMultipartDownloaderSubscriber
      * attempted for that part, but it still failed. This is a failure state, the error should be reported back to the user
      * and any more request should be ignored.
      */
-    private final AtomicBoolean completedExceptionally = new AtomicBoolean(false);
+    private final AtomicBoolean isCompletedExceptionally = new AtomicBoolean(false);
 
     public NonLinearMultipartDownloaderSubscriber(S3AsyncClient s3,
                                                   GetObjectRequest getObjectRequest,
@@ -185,7 +185,7 @@ public void onNext(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse
 
     private void executeRequestOrAddToPending(
         AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
-        if (handleFirstRequest(asyncResponseTransformer)) {
+        if (handleFirstRequestOrEnqueueTransformer(asyncResponseTransformer)) {
             return;
         }
 
@@ -205,16 +205,16 @@ private void executeRequestOrAddToPending(
      * We need to wait for the first request to finish so we know if it is aa multipart object or not.
      * While we don't know yet, additional onNext signal receives are stored in pendingTransformers.
      */
-    private boolean handleFirstRequest(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
+    private boolean handleFirstRequestOrEnqueueTransformer(AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> asyncResponseTransformer) {
         if (completedParts.get() != 0) {
             return false;
         }
 
         firstPartLock.lock();
         try {
-            if (!firstRequestSent.get()) {
+            if (!isFirstRequestSent.get()) {
                 sendFirstRequest(asyncResponseTransformer);
-                firstRequestSent.set(true);
+                isFirstRequestSent.set(true);
                 return true;
             }
 
@@ -251,7 +251,7 @@ private void sendNextRequest(AsyncResponseTransformer<GetObjectResponse, GetObje
             inFlightRequests.remove(partToGet);
 
             completedParts.incrementAndGet();
-            if (e != null || completedExceptionally.get()) {
+            if (e != null || isCompletedExceptionally.get()) {
                 // Note on retries: When this future completes exceptionally, it means we did all retries and still failed for
                 // that part. We need to report back the failure to the user.
                 handlePartError(e, partToGet);
@@ -286,7 +286,7 @@ private void sendFirstRequest(AsyncResponseTransformer<GetObjectResponse, GetObj
             inFlightRequests.remove(1);
 
             completedParts.incrementAndGet();
-            if (e != null || completedExceptionally.get()) {
+            if (e != null || isCompletedExceptionally.get()) {
                 // Note on retries: When this future completes exceptionally, it means we did all retries and still failed for
                 // that part. We need to report back the failure to the user.
                 handlePartError(e, 1);
@@ -337,7 +337,7 @@ private void setInitialPartCountAndEtag(GetObjectResponse response) {
     }
 
     private void handlePartError(Throwable e, int part) {
-        completedExceptionally.set(true);
+        isCompletedExceptionally.set(true);
         log.error(() -> "Error on part " + part + ": " + e);
         resultFuture.completeExceptionally(e);
         inFlightRequests.values().forEach(future -> future.cancel(true));