Return as soon as the stream starts if there is a payload in sync http client

Author: zoewangg

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

@@ -67,8 +67,9 @@ final class ClasspathSdkHttpServiceProvider<T> implements SdkHttpServiceProvider
 
     @Override
     public Optional<T> loadService() {
-        Queue<T> impls = new PriorityQueue<>(Comparator.comparingInt(o -> httpServicesPriority.getOrDefault(o.getClass().getName(),
-                                                                                                            Integer.MAX_VALUE)));
+        Queue<T> impls = new PriorityQueue<>(
+            Comparator.comparingInt(o -> httpServicesPriority.getOrDefault(o.getClass().getName(),
+                                                                           Integer.MAX_VALUE)));
         Iterable<T> iterable = () -> serviceLoader.loadServices(serviceClass);
         iterable.forEach(impl -> impls.add(impl));
 

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/response/InputStreamAdaptingHttpStreamResponseHandler.java

@@ -69,7 +69,10 @@ public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
         if (inputStreamSubscriber == null) {
             inputStreamSubscriber = new InputStreamSubscriber();
             simplePublisher.subscribe(inputStreamSubscriber);
+            // For response with a payload, we need to complete the future here to allow downstream to retrieve the data from
+            // the stream directly.
             responseBuilder.content(AbortableInputStream.create(inputStreamSubscriber));
+            requestCompletionFuture.complete(responseBuilder.build());
         }
 
         CompletableFuture<Void> writeFuture = simplePublisher.send(ByteBuffer.wrap(bodyBytesIn));
@@ -123,6 +126,8 @@ private void onSuccessfulResponseComplete(HttpStream stream) {
             crtConn.shutdown();
         }
 
+        // For response without a payload, for example, S3 PutObjectResponse, we need to complete the future
+        // in onResponseComplete callback since onResponseBody will never be invoked.
         requestCompletionFuture.complete(responseBuilder.build());
         simplePublisher.complete();
         crtConn.close();