Hopefully fix the build this time.

Author: JonathanHenson

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

@@ -132,6 +132,12 @@ protected void request(long n) {
             outstandingReqs = outstandingRequests.addAndGet(n);
         }
 
+        /*
+         * Since we buffer, in the case where the subscriber came in after the publication has already begun,
+         * go ahead and flush what we have.
+         */
+        publishToSubscribers();
+
         log.trace(() -> "Subscriber Requested more Buffers. Outstanding Requests: " + outstandingReqs);
     }
 
@@ -249,7 +255,7 @@ protected synchronized void publishToSubscribers() {
         }
 
         // Check if Complete, consider no subscriber as a completion.
-        if (queueComplete.get() && (subscriberRef.get() == null || queuedBuffers.size() == 0)) {
+        if (queueComplete.get() && queuedBuffers.size() == 0) {
             completeSubscriptionExactlyOnce();
         }
     }

http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtHttpClientSpiVerificationTest.java

@@ -201,13 +201,33 @@ private void makePutRequest(String path, byte[] reqBody, int expectedStatus) thr
         final AtomicReference<SdkHttpResponse> response = new AtomicReference<>(null);
         final AtomicReference<Throwable> error = new AtomicReference<>(null);
 
+        Subscriber<ByteBuffer> subscriber = new Subscriber<ByteBuffer>() {
+            @Override
+            public void onSubscribe(Subscription subscription) {
+                subscription.request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onNext(ByteBuffer byteBuffer) {
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+            }
+
+            @Override
+            public void onComplete() {
+            }
+        };
+
         SdkAsyncHttpResponseHandler handler = new SdkAsyncHttpResponseHandler() {
             @Override
             public void onHeaders(SdkHttpResponse headers) {
                 response.compareAndSet(null, headers);
             }
             @Override
             public void onStream(Publisher<ByteBuffer> stream) {
+                stream.subscribe(subscriber);
                 streamReceived.complete(true);
             }
 

test/sdk-benchmarks/src/main/java/software/amazon/awssdk/benchmark/apicall/httpclient/async/AwsCrtClientNonTLSBenchmark.java

@@ -15,7 +15,25 @@
 
 package software.amazon.awssdk.benchmark.apicall.httpclient.async;
 
-import org.openjdk.jmh.annotations.*;
+import static software.amazon.awssdk.benchmark.utils.BenchmarkConstant.CONCURRENT_CALLS;
+import static software.amazon.awssdk.benchmark.utils.BenchmarkUtils.awaitCountdownLatchUninterruptibly;
+import static software.amazon.awssdk.benchmark.utils.BenchmarkUtils.countDownUponCompletion;
+
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OperationsPerInvocation;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
 import org.openjdk.jmh.infra.Blackhole;
 import org.openjdk.jmh.profile.StackProfiler;
 import org.openjdk.jmh.results.RunResult;
@@ -28,14 +46,6 @@
 import software.amazon.awssdk.http.crt.AwsCrtAsyncHttpClient;
 import software.amazon.awssdk.services.protocolrestjson.ProtocolRestJsonAsyncClient;
 
-import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import static software.amazon.awssdk.benchmark.utils.BenchmarkConstant.CONCURRENT_CALLS;
-import static software.amazon.awssdk.benchmark.utils.BenchmarkUtils.awaitCountdownLatchUninterruptibly;
-import static software.amazon.awssdk.benchmark.utils.BenchmarkUtils.countDownUponCompletion;
-
 /**
  * Using aws-crt-client to test against local mock https server.
  */