Remove default wrapWithListener methods & make SdkProtectedApi

Author: Bennett Lynch

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

@@ -22,7 +22,6 @@
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.core.ResponseBytes;
 import software.amazon.awssdk.core.SdkResponse;
-import software.amazon.awssdk.core.async.listen.AsyncResponseTransformerListener;
 import software.amazon.awssdk.core.internal.async.ByteArrayAsyncResponseTransformer;
 import software.amazon.awssdk.core.internal.async.FileAsyncResponseTransformer;
 import software.amazon.awssdk.core.internal.async.PublisherAsyncResponseTransformer;
@@ -109,14 +108,6 @@ public interface AsyncResponseTransformer<ResponseT, ResultT> {
      */
     void exceptionOccurred(Throwable error);
 
-    /**
-     * Wrap this {@link AsyncResponseTransformer} with a new one that will notify a {@link AsyncResponseTransformerListener} of
-     * important events occurring.
-     */
-    default AsyncResponseTransformer<ResponseT, ResultT> wrapWithListener(AsyncResponseTransformerListener<ResponseT> listener) {
-        return AsyncResponseTransformerListener.wrap(this, listener);
-    }
-    
     /**
      * Creates an {@link AsyncResponseTransformer} that writes all the content to the given file. In the event of an error,
      * the SDK will attempt to delete the file (whatever has been written to it so far). If the file already exists, an

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

@@ -117,15 +117,15 @@ default SdkPublisher<List<T>> buffer(int bufferSize) {
     default SdkPublisher<T> limit(int limit) {
         return subscriber -> subscribe(new LimitingSubscriber<>(subscriber, limit));
     }
-    
+
     /**
      * Add a callback that will be invoked after this publisher invokes {@link Subscriber#onComplete()}.
      *
      * @param afterOnComplete The logic that should be run immediately after onComplete.
      * @return New publisher that invokes the requested callback.
      */
     default SdkPublisher<T> doAfterOnComplete(Runnable afterOnComplete) {
-        return wrapWithListener(new PublisherListener<T>() {
+        return PublisherListener.wrap(this, new PublisherListener<T>() {
             @Override
             public void subscriberOnComplete() {
                 afterOnComplete.run();
@@ -140,7 +140,7 @@ public void subscriberOnComplete() {
      * @return New publisher that invokes the requested callback.
      */
     default SdkPublisher<T> doAfterOnError(Consumer<Throwable> afterOnError) {
-        return wrapWithListener(new PublisherListener<T>() {
+        return PublisherListener.wrap(this, new PublisherListener<T>() {
             @Override
             public void subscriberOnError(Throwable t) {
                 afterOnError.accept(t);
@@ -155,7 +155,7 @@ public void subscriberOnError(Throwable t) {
      * @return New publisher that invokes the requested callback.
      */
     default SdkPublisher<T> doAfterOnCancel(Runnable afterOnCancel) {
-        return wrapWithListener(new PublisherListener<T>() {
+        return PublisherListener.wrap(this, new PublisherListener<T>() {
             @Override
             public void subscriptionCancel() {
                 afterOnCancel.run();
@@ -177,10 +177,4 @@ default CompletableFuture<Void> subscribe(Consumer<T> consumer) {
         return future;
     }
 
-    /**
-     * Wrap this {@link SdkPublisher} with a new one that will notify a {@link PublisherListener} of important events occurring.
-     */
-    default SdkPublisher<T> wrapWithListener(PublisherListener<T> listener) {
-        return PublisherListener.wrap(this, listener);
-    }
 }

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

@@ -29,12 +29,13 @@
 import software.amazon.awssdk.utils.Validate;
 
 /**
- * Listener interface that invokes callbacks associated with this {@link AsyncResponseTransformer} and any resulting {@link
+ * Listener interface that invokes callbacks associated with a {@link AsyncResponseTransformer} and any resulting {@link
  * SdkPublisher} and {@link Subscriber}.
  *
  * @see PublisherListener
  * @see SubscriberListener
  */
+@SdkProtectedApi
 public interface AsyncResponseTransformerListener<ResponseT> extends PublisherListener<ByteBuffer> {
 
     /**

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

@@ -28,11 +28,12 @@
 import software.amazon.awssdk.utils.Validate;
 
 /**
- * Listener interface that invokes callbacks associated with this {@link Publisher} and any resulting {@link Subscriber}.
+ * Listener interface that invokes callbacks associated with a {@link Publisher} and any resulting {@link Subscriber}.
  *
  * @see AsyncResponseTransformerListener
  * @see SubscriberListener
  */
+@SdkProtectedApi
 public interface PublisherListener<T> extends SubscriberListener<T> {
     /**
      * Invoked after {@link Publisher#subscribe(Subscriber)}
@@ -41,7 +42,7 @@ default void publisherSubscribe(Subscriber<? super T> subscriber) {
     }
 
     /**
-     * Wrap this {@link SdkPublisher} with a new one that will notify a {@link PublisherListener} of important events occurring.
+     * Wrap a {@link SdkPublisher} with a new one that will notify a {@link PublisherListener} of important events occurring.
      */
     static <T> SdkPublisher<T> wrap(SdkPublisher<T> delegate, PublisherListener<T> listener) {
         return new NotifyingPublisher<>(delegate, listener);

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

@@ -26,11 +26,12 @@
 import software.amazon.awssdk.utils.Validate;
 
 /**
- * Listener interface that invokes callbacks associated with this {@link Subscriber}.
+ * Listener interface that invokes callbacks associated with a {@link Subscriber}.
  *
  * @see AsyncResponseTransformerListener
  * @see PublisherListener
  */
+@SdkProtectedApi
 public interface SubscriberListener<T> {
     /**
      * Invoked after {@link Subscriber#onNext(Object)}
@@ -57,7 +58,7 @@ default void subscriptionCancel() {
     }
 
     /**
-     * Wrap this {@link Subscriber} with a new one that will notify a {@link SubscriberListener} of important events occurring.
+     * Wrap a {@link Subscriber} with a new one that will notify a {@link SubscriberListener} of important events occurring.
      */
     static <T> Subscriber<T> wrap(Subscriber<? super T> delegate, SubscriberListener<? super T> listener) {
         return new NotifyingSubscriber<>(delegate, listener);
@@ -108,7 +109,7 @@ final class NotifyingSubscription implements Subscription {
             private final Subscription delegateSubscription;
 
             NotifyingSubscription(Subscription delegateSubscription) {
-                this.delegateSubscription = Validate.notNull(delegateSubscription, "delegate");
+                this.delegateSubscription = Validate.notNull(delegateSubscription, "delegateSubscription");
             }
 
             @Override

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

@@ -31,22 +31,24 @@ public class TmpUtil {
     public static <A, B> Pair<AsyncResponseTransformer<A, B>, CompletableFuture<Void>> wrapWithEndOfStreamFuture(
         AsyncResponseTransformer<A, B> responseTransformer) {
         CompletableFuture<Void> future = new CompletableFuture<>();
-        AsyncResponseTransformer<A, B> wrapped = responseTransformer.wrapWithListener(new AsyncResponseTransformerListener<A>() {
-            @Override
-            public void transformerExceptionOccurred(Throwable t) {
-                future.completeExceptionally(t);
-            }
+        AsyncResponseTransformer<A, B> wrapped = AsyncResponseTransformerListener.wrap(
+            responseTransformer,
+            new AsyncResponseTransformerListener<A>() {
+                @Override
+                public void transformerExceptionOccurred(Throwable t) {
+                    future.completeExceptionally(t);
+                }
 
-            @Override
-            public void subscriberOnError(Throwable t) {
-                future.completeExceptionally(t);
-            }
+                @Override
+                public void subscriberOnError(Throwable t) {
+                    future.completeExceptionally(t);
+                }
 
-            @Override
-            public void subscriberOnComplete() {
-                future.complete(null);
-            }
-        });
+                @Override
+                public void subscriberOnComplete() {
+                    future.complete(null);
+                }
+            });
         return Pair.of(wrapped, future);
     }
 }

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/progress/TransferProgressUpdater.java

@@ -87,45 +87,47 @@ public void beforeOnNext(ByteBuffer byteBuffer) {
 
     public <ResultT> AsyncResponseTransformer<GetObjectResponse, ResultT> wrapResponseTransformer(
         AsyncResponseTransformer<GetObjectResponse, ResultT> responseTransformer) {
-        return responseTransformer.wrapWithListener(new AsyncResponseTransformerListener<GetObjectResponse>() {
-            @Override
-            public void transformerOnResponse(GetObjectResponse response) {
-                if (response.contentLength() != null) {
-                    progress.updateAndGet(b -> b.transferSizeInBytes(response.contentLength()));
+        return AsyncResponseTransformerListener.wrap(
+            responseTransformer,
+            new AsyncResponseTransformerListener<GetObjectResponse>() {
+                @Override
+                public void transformerOnResponse(GetObjectResponse response) {
+                    if (response.contentLength() != null) {
+                        progress.updateAndGet(b -> b.transferSizeInBytes(response.contentLength()));
+                    }
                 }
-            }
 
-            @Override
-            public void transformerExceptionOccurred(Throwable t) {
-                transferFailed(t);
-            }
+                @Override
+                public void transformerExceptionOccurred(Throwable t) {
+                    transferFailed(t);
+                }
 
-            @Override
-            public void publisherSubscribe(Subscriber<? super ByteBuffer> subscriber) {
-                progress.updateAndGet(b -> b.bytesTransferred(0));
-            }
+                @Override
+                public void publisherSubscribe(Subscriber<? super ByteBuffer> subscriber) {
+                    progress.updateAndGet(b -> b.bytesTransferred(0));
+                }
 
-            @Override
-            public void subscriberOnNext(ByteBuffer byteBuffer) {
-                TransferProgressSnapshot snapshot = progress.updateAndGet(b -> {
-                    b.bytesTransferred(b.getBytesTransferred() + byteBuffer.limit());
-                });
-                listeners.bytesTransferred(context.copy(b -> b.progressSnapshot(snapshot)));
-            }
+                @Override
+                public void subscriberOnNext(ByteBuffer byteBuffer) {
+                    TransferProgressSnapshot snapshot = progress.updateAndGet(b -> {
+                        b.bytesTransferred(b.getBytesTransferred() + byteBuffer.limit());
+                    });
+                    listeners.bytesTransferred(context.copy(b -> b.progressSnapshot(snapshot)));
+                }
 
-            @Override
-            public void subscriberOnError(Throwable t) {
-                transferFailed(t);
-            }
+                @Override
+                public void subscriberOnError(Throwable t) {
+                    transferFailed(t);
+                }
 
-            @Override
-            public void subscriberOnComplete() {
-                listeners.transferComplete(context.copy(b -> {
-                    b.progressSnapshot(progress.snapshot());
-                    b.completedTransfer(completedTransfer);
-                }));
-            }
-        });
+                @Override
+                public void subscriberOnComplete() {
+                    listeners.transferComplete(context.copy(b -> {
+                        b.progressSnapshot(progress.snapshot());
+                        b.completedTransfer(completedTransfer);
+                    }));
+                }
+            });
     }
 
     public void registerCompletion(CompletableFuture<? extends CompletedObjectTransfer> future) {