Compliance with Reactive Streams specs

Author: davidh44

utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java

@@ -101,6 +101,12 @@ public final class SimplePublisher<T> implements Publisher<T> {
      */
     private final AtomicBoolean subscribed = new AtomicBoolean(false);
 
+    /**
+     * True while the subscriber's {@code onSubscribe} is executing. Used to prevent {@link #processEventQueue()} from
+     * delivering {@code onNext} signals before {@code onSubscribe} has returned, as required by Reactive Streams rule 1.03.
+     */
+    private volatile boolean onSubscribeInProgress = false;
+
     /**
      * The subscriber provided via {@link #subscribe(Subscriber)}. This publisher only supports a single subscriber.
      *
@@ -212,7 +218,12 @@ public void subscribe(Subscriber<? super T> s) {
         }
 
         this.subscriber = s;
-        s.onSubscribe(new SubscriptionImpl());
+        onSubscribeInProgress = true;
+        try {
+            s.onSubscribe(new SubscriptionImpl());
+        } finally {
+            onSubscribeInProgress = false;
+        }
         processEventQueue();
     }
 
@@ -329,6 +340,11 @@ private boolean shouldProcessQueueEntry(QueueEntry<T> entry) {
             return false;
         }
 
+        if (onSubscribeInProgress) {
+            // Do not deliver signals until onSubscribe has returned (Reactive Streams rule 1.03).
+            return false;
+        }
+
         if (entry.type() != ON_NEXT) {
             // This event isn't an on-next event, so we don't need subscriber demand in order to process it.
             return true;