Add Progress Listener Invocation methods to Asynchronous and Synchronous code paths

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonAsyncHttpClient.java

@@ -46,6 +46,8 @@
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeRequestMutableStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomHeadersStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomQueryParamsStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.PostExecutionUpdateProgressStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.PreExecutionUpdateProgressStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.QueryParametersToBodyStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.UnwrapResponseContainer;
 import software.amazon.awssdk.core.internal.util.ThrowableUtils;
@@ -201,12 +203,14 @@ public <OutputT> CompletableFuture<OutputT> execute(
                                 .then(() -> new HttpChecksumStage(ClientType.ASYNC))
                                 .then(MakeRequestImmutableStage::new)
                                 .then(RequestPipelineBuilder
-                                        .first(AsyncSigningStage::new)
+                                        .first(PreExecutionUpdateProgressStage::new)
+                                        .then(AsyncSigningStage::new)
                                         .then(AsyncBeforeTransmissionExecutionInterceptorsStage::new)
                                         .then(d -> new MakeAsyncHttpRequestStage<>(responseHandler, d))
                                         .wrappedWith(AsyncApiCallAttemptMetricCollectionStage::new)
                                         .wrappedWith((deps, wrapped) -> new AsyncRetryableStage<>(responseHandler, deps, wrapped))
                                         .then(async(() -> new UnwrapResponseContainer<>()))
+                                        .then(() -> new PostExecutionUpdateProgressStage<>())
                                         .then(async(() -> new AfterExecutionInterceptorsStage<>()))
                                         .wrappedWith(AsyncExecutionFailureExceptionReportingStage::new)
                                         .wrappedWith(AsyncApiCallTimeoutTrackingStage::new)

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonSyncHttpClient.java

@@ -46,6 +46,8 @@
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeRequestMutableStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomHeadersStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MergeCustomQueryParamsStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.PostExecutionUpdateProgressStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.PreExecutionUpdateProgressStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.QueryParametersToBodyStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.SigningStage;
@@ -204,7 +206,8 @@ public <OutputT> OutputT execute(HttpResponseHandler<Response<OutputT>> response
                                .then(MakeRequestImmutableStage::new)
                                // End of mutating request
                                .then(RequestPipelineBuilder
-                                         .first(SigningStage::new)
+                                         .first(PreExecutionUpdateProgressStage::new)
+                                         .then(SigningStage::new)
                                          .then(BeforeTransmissionExecutionInterceptorsStage::new)
                                          .then(MakeHttpRequestStage::new)
                                          .then(AfterTransmissionExecutionInterceptorsStage::new)
@@ -219,6 +222,7 @@ public <OutputT> OutputT execute(HttpResponseHandler<Response<OutputT>> response
                                .wrappedWith((deps, wrapped) -> new ApiCallMetricCollectionStage<>(wrapped))
                     .then(() -> new UnwrapResponseContainer<>())
                     .then(() -> new AfterExecutionInterceptorsStage<>())
+                    .then(() -> new PostExecutionUpdateProgressStage<>())
                     .wrappedWith(ExecutionFailureExceptionReportingStage::new)
                     .build(httpClientDependencies)
                     .execute(request, createRequestExecutionDependencies());

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncExecutionFailureExceptionReportingStage.java

@@ -42,12 +42,15 @@ public CompletableFuture<OutputT> execute(SdkHttpFullRequest input, RequestExecu
         CompletableFuture<OutputT> executeFuture = wrappedExecute.handle((o, t) -> {
             if (t != null) {
                 Throwable toReport = t;
-
                 if (toReport instanceof CompletionException) {
                     toReport = toReport.getCause();
                 }
                 toReport = reportFailureToInterceptors(context, toReport);
 
+                context.executionContext().progressUpdater().ifPresent(progressUpdater -> {
+                    progressUpdater.attemptFailure(t);
+                });
+
                 throw CompletableFutureUtils.errorAsCompletionException(ThrowableUtils.asSdkException(toReport));
             } else {
                 return o;

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/ExecutionFailureExceptionReportingStage.java

@@ -37,6 +37,10 @@ public OutputT execute(SdkHttpFullRequest input, RequestExecutionContext context
             return wrapped.execute(input, context);
         } catch (Exception e) {
             Throwable throwable = reportFailureToInterceptors(context, e);
+
+            context.executionContext().progressUpdater().ifPresent(progressUpdater -> {
+                progressUpdater.attemptFailure(e);
+            });
             throw failure(throwable);
         }
     }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/MakeAsyncHttpRequestStage.java

@@ -47,6 +47,8 @@
 import software.amazon.awssdk.core.internal.http.timers.TimeoutTracker;
 import software.amazon.awssdk.core.internal.http.timers.TimerUtils;
 import software.amazon.awssdk.core.internal.metrics.BytesReadTrackingPublisher;
+import software.amazon.awssdk.core.internal.metrics.BytesSentTrackingPublisher;
+import software.amazon.awssdk.core.internal.progress.listener.ProgressUpdater;
 import software.amazon.awssdk.core.internal.util.MetricUtils;
 import software.amazon.awssdk.core.metrics.CoreMetric;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
@@ -58,6 +60,7 @@
 import software.amazon.awssdk.metrics.MetricCollector;
 import software.amazon.awssdk.utils.CompletableFutureUtils;
 import software.amazon.awssdk.utils.Logger;
+import software.amazon.awssdk.utils.StringUtils;
 
 /**
  * Delegate to the HTTP implementation to make an HTTP request and receive the response.
@@ -137,6 +140,21 @@ private CompletableFuture<Response<OutputT>> executeHttpRequest(SdkHttpFullReque
 
         MetricCollector httpMetricCollector = MetricUtils.createHttpMetricsCollector(context);
 
+        //If Progress Listening is enabled, wrap around BytesSentTrackingPublisher to track progress on bytes sent
+        if (context.executionContext().progressUpdater().isPresent()) {
+
+            ProgressUpdater progressUpdater = context.executionContext().progressUpdater().get();
+
+            if (shouldSetContentLength(request, requestProvider)) {
+                progressUpdater.updateRequestContentLength(requestProvider.contentLength().get());
+            }
+
+            requestProvider = new BytesSentTrackingPublisher(requestProvider,
+                                                             progressUpdater,
+                                                             requestProvider.contentLength());
+
+        }
+
         AsyncExecuteRequest.Builder executeRequestBuilder = AsyncExecuteRequest.builder()
                                                                 .request(requestWithContentLength)
                                                                 .requestContentPublisher(requestProvider)
@@ -303,13 +321,27 @@ public void onHeaders(SdkHttpResponse headers) {
             long d = now - startTime;
             context.attemptMetricCollector().reportMetric(CoreMetric.TIME_TO_FIRST_BYTE, Duration.ofNanos(d));
             super.onHeaders(headers);
+
+            context.executionContext().progressUpdater().ifPresent(progressUpdater -> {
+                progressUpdater.responseHeaderReceived();
+                headers.firstMatchingHeader(CONTENT_LENGTH).ifPresent(value -> {
+                    if (!StringUtils.isNotBlank(value)) {
+                        progressUpdater.updateResponseContentLength(Long.parseLong(value));
+                    }
+                });
+            });
         }
 
         @Override
         public void onStream(Publisher<ByteBuffer> stream) {
+            BytesReadTrackingPublisher bytesReadTrackingPublisher;
             AtomicLong bytesReadCounter = context.executionAttributes()
                                                  .getAttribute(SdkInternalExecutionAttribute.RESPONSE_BYTES_READ);
-            BytesReadTrackingPublisher bytesReadTrackingPublisher = new BytesReadTrackingPublisher(stream, bytesReadCounter);
+
+            bytesReadTrackingPublisher = new BytesReadTrackingPublisher(stream,
+                                                                        bytesReadCounter,
+                                                                        context.executionContext().progressUpdater());
+
             super.onStream(bytesReadTrackingPublisher);
         }
     }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/PostExecutionUpdateProgressStage.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.http.pipeline.stages;
+
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.SdkResponse;
+import software.amazon.awssdk.core.internal.http.RequestExecutionContext;
+import software.amazon.awssdk.core.internal.http.pipeline.RequestPipeline;
+
+@SdkInternalApi
+public class PostExecutionUpdateProgressStage<OutputT> implements RequestPipeline<OutputT, OutputT> {
+    @Override
+    public OutputT execute(OutputT input, RequestExecutionContext context) throws Exception {
+
+        context.executionContext().progressUpdater().ifPresent(progressUpdater -> {
+            progressUpdater.executionSuccess((SdkResponse) input);
+        });
+        return input;
+    }
+
+
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/PreExecutionUpdateProgressStage.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.core.internal.http.pipeline.stages;
+
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.internal.http.RequestExecutionContext;
+import software.amazon.awssdk.core.internal.http.pipeline.RequestToRequestPipeline;
+import software.amazon.awssdk.core.internal.progress.listener.ProgressUpdater;
+import software.amazon.awssdk.http.SdkHttpFullRequest;
+
+@SdkInternalApi
+public class PreExecutionUpdateProgressStage implements RequestToRequestPipeline {
+
+    @Override
+    public SdkHttpFullRequest execute(SdkHttpFullRequest input, RequestExecutionContext context) throws Exception {
+
+        if (progressListenerAttached(context.originalRequest())) {
+            Long requestContentLength =
+                context.requestProvider() != null && context.requestProvider().contentLength().isPresent() ?
+                context.requestProvider().contentLength().get() : null;
+
+            if (context.executionContext().progressUpdater().isPresent()) {
+                context.executionContext().progressUpdater().get().requestPrepared(input);
+            } else {
+                ProgressUpdater progressUpdater = new ProgressUpdater(context.originalRequest(), requestContentLength);
+                progressUpdater.requestPrepared(input);
+                context.executionContext().toBuilder().progressUpdater(progressUpdater);
+            }
+        }
+
+        return input;
+    }
+
+    public boolean progressListenerAttached(SdkRequest request) {
+        if (request.overrideConfiguration().isPresent() &&
+            !request.overrideConfiguration().get().progressListeners().isEmpty()) {
+            return true;
+        }
+        return false;
+    }
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/UnwrapResponseContainer.java

@@ -17,6 +17,7 @@
 
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.core.Response;
+import software.amazon.awssdk.core.SdkResponse;
 import software.amazon.awssdk.core.internal.http.RequestExecutionContext;
 import software.amazon.awssdk.core.internal.http.pipeline.RequestPipeline;
 
@@ -31,6 +32,9 @@
 public class UnwrapResponseContainer<OutputT> implements RequestPipeline<Response<OutputT>, OutputT> {
     @Override
     public OutputT execute(Response<OutputT> input, RequestExecutionContext context) throws Exception {
+        context.executionContext().progressUpdater().ifPresent(progressUpdater -> {
+            progressUpdater.executionSuccess((SdkResponse) input.response());
+        });
         return input.response();
     }
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/metrics/BytesReadTrackingPublisher.java

@@ -16,11 +16,13 @@
 package software.amazon.awssdk.core.internal.metrics;
 
 import java.nio.ByteBuffer;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.internal.progress.listener.ProgressUpdater;
 
 /**
  * Publisher that tracks how many bytes are published from the wrapped publisher to the downstream subscriber.
@@ -29,15 +31,20 @@
 public final class BytesReadTrackingPublisher implements Publisher<ByteBuffer> {
     private final Publisher<ByteBuffer> upstream;
     private final AtomicLong bytesRead;
+    private ProgressUpdater progressUpdater;
 
-    public BytesReadTrackingPublisher(Publisher<ByteBuffer> upstream, AtomicLong bytesRead) {
+    public BytesReadTrackingPublisher(Publisher<ByteBuffer> upstream, AtomicLong bytesRead,
+                                      Optional<ProgressUpdater> progressUpdater) {
         this.upstream = upstream;
         this.bytesRead = bytesRead;
+        progressUpdater.ifPresent(value -> {
+            this.progressUpdater = value;
+        });
     }
 
     @Override
     public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
-        upstream.subscribe(new BytesReadTracker(subscriber, bytesRead));
+        upstream.subscribe(new BytesReadTracker(subscriber, bytesRead, progressUpdater));
     }
 
     public long bytesRead() {
@@ -47,21 +54,30 @@ public long bytesRead() {
     private static final class BytesReadTracker implements Subscriber<ByteBuffer> {
         private final Subscriber<? super ByteBuffer> downstream;
         private final AtomicLong bytesRead;
+        private final ProgressUpdater progressUpdater;
 
-        private BytesReadTracker(Subscriber<? super ByteBuffer> downstream, AtomicLong bytesRead) {
+        private BytesReadTracker(Subscriber<? super ByteBuffer> downstream,
+                                 AtomicLong bytesRead, ProgressUpdater progressUpdater) {
             this.downstream = downstream;
             this.bytesRead = bytesRead;
+            this.progressUpdater = progressUpdater;
         }
 
         @Override
         public void onSubscribe(Subscription subscription) {
             downstream.onSubscribe(subscription);
+            if (progressUpdater != null) {
+                progressUpdater.resetBytesReceived();
+            }
         }
 
         @Override
         public void onNext(ByteBuffer byteBuffer) {
             bytesRead.addAndGet(byteBuffer.remaining());
             downstream.onNext(byteBuffer);
+            if (progressUpdater != null) {
+                progressUpdater.incrementBytesReceived(bytesRead.get());
+            }
         }
 
         @Override