Fixed an issue in JSON-RPC event stream services (none exist yet) where retries excluded the initial-request event. (#2673)

Before this change, the AsyncStreamPrepender (a Publisher) does not support multiple subscribe calls, but each attempt makes
a new subscribe call. Because of this, if a retry is encountered, backpressure is miscalculated and the initial-request
is not sent for the retries.

Author: millems

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

@@ -25,9 +25,6 @@
 public class AsyncStreamPrepender<T> implements Publisher<T> {
     private final Publisher<T> delegate;
     private final T firstItem;
-    private Subscriber<? super T> subscriber;
-    private volatile boolean complete = false;
-    private volatile boolean firstRequest = true;
 
     public AsyncStreamPrepender(Publisher<T> delegate, T firstItem) {
         this.delegate = delegate;
@@ -36,11 +33,18 @@ public AsyncStreamPrepender(Publisher<T> delegate, T firstItem) {
 
     @Override
     public void subscribe(Subscriber<? super T> s) {
-        subscriber = s;
-        delegate.subscribe(new DelegateSubscriber());
+        delegate.subscribe(new DelegateSubscriber(s));
     }
 
     private class DelegateSubscriber implements Subscriber<T> {
+        private final Subscriber<? super T> subscriber;
+        private volatile boolean complete = false;
+        private volatile boolean firstRequest = true;
+
+        private DelegateSubscriber(Subscriber<? super T> subscriber) {
+            this.subscriber = subscriber;
+        }
+
         @Override
         public void onSubscribe(Subscription subscription) {
             subscriber.onSubscribe(new Subscription() {
@@ -90,7 +94,6 @@ public void request(long n) {
                 public void cancel() {
                     cancelled = true;
                     subscription.cancel();
-                    subscriber = null;
                 }
             });
         }

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/AsyncStreamPrependerTest.java

@@ -62,6 +62,21 @@ public void sequence() {
         }
     }
 
+    @Test
+    public void multiSubscribe_stillPrepends() {
+        AsyncStreamPrepender<Long> prepender = new AsyncStreamPrepender<>(rangeLong(1L, 5L), 0L);
+        Flowable<Long> prepended1 = fromPublisher(prepender);
+        Flowable<Long> prepended2 = fromPublisher(prepender);
+
+        Iterator<Long> iterator1 = prepended1.blockingIterable(1).iterator();
+        Iterator<Long> iterator2 = prepended2.blockingIterable(1).iterator();
+
+        for (long i = 0; i <= 5; i++) {
+            assertEquals(i, iterator1.next().longValue());
+            assertEquals(i, iterator2.next().longValue());
+        }
+    }
+
     @Test
     public void error() {
         Flowable<Long> error = Flowable.error(IllegalStateException::new);