review reactive-streams#1

cancelled being true here is impossible because the only way to get that
true is by submitting a Subscription to the Subscriber that happens once
and under this condition.

Author: ttulka

examples/src/main/java/org/reactivestreams/example/unicast/AsyncStreamPublisher.java

@@ -126,38 +126,36 @@ private void doCancel() {
         }
 
         // Instead of executing `subscriber.onSubscribe` synchronously from within `Publisher.subscribe`
-        // we execute it asynchronously, this is to avoid executing the user code (`Iterable.iterator`) on the calling thread.
+        // we execute it asynchronously, this is to avoid executing the user code (`supplier`) on the calling thread.
         // It also makes it easier to follow rule 1.9
         private void doSubscribe() {
-            if (!cancelled) {
-                // Deal with setting up the subscription with the subscriber
-                try {
-                    subscriber.onSubscribe(this);
-                } catch (final Throwable t) { // Due diligence to obey 2.13
-                    terminateDueTo(new IllegalStateException(subscriber + " violated the Reactive Streams rule 2.13 by throwing an exception from onSubscribe.", t));
-                }
+            // Deal with setting up the subscription with the subscriber
+            try {
+                subscriber.onSubscribe(this);
+            } catch (final Throwable t) { // Due diligence to obey 2.13
+                terminateDueTo(new IllegalStateException(subscriber + " violated the Reactive Streams rule 2.13 by throwing an exception from onSubscribe.", t));
+            }
 
-                // Deal with already complete iterators promptly
-                boolean hasElements = false;
-                try {
-                    // Try to fetch an element from a stream to ensure the stream is not empty,
-                    // this will be sent by the first calling of doSend
-                    nextElementToBeSent = supplier.get();
-                    hasElements = nextElementToBeSent != null;
-                } catch (final Throwable t) {
-                    terminateDueTo(t); // If hasNext throws, there's something wrong and we need to signal onError as per 1.2, 1.4
-                    return;
-                }
+            // Deal with already complete iterators promptly
+            boolean hasElements = false;
+            try {
+                // Try to fetch an element from a stream to ensure the stream is not empty,
+                // this will be sent by the first calling of doSend
+                nextElementToBeSent = supplier.get();
+                hasElements = nextElementToBeSent != null;
+            } catch (final Throwable t) {
+                terminateDueTo(t); // If hasNext throws, there's something wrong and we need to signal onError as per 1.2, 1.4
+                return;
+            }
 
-                // If we don't have anything to deliver, we're already done, so lets do the right thing and
-                // not wait for demand to deliver `onComplete` as per rule 1.2 and 1.3
-                if (!hasElements) {
-                    try {
-                        doCancel(); // Rule 1.6 says we need to consider the `Subscription` cancelled when `onComplete` is signalled
-                        subscriber.onComplete();
-                    } catch (final Throwable t) { // As per rule 2.13, `onComplete` is not allowed to throw exceptions, so we do what we can, and log this.
-                        (new IllegalStateException(subscriber + " violated the Reactive Streams rule 2.13 by throwing an exception from onComplete.", t)).printStackTrace(System.err);
-                    }
+            // If we don't have anything to deliver, we're already done, so lets do the right thing and
+            // not wait for demand to deliver `onComplete` as per rule 1.2 and 1.3
+            if (!hasElements) {
+                try {
+                    doCancel(); // Rule 1.6 says we need to consider the `Subscription` cancelled when `onComplete` is signalled
+                    subscriber.onComplete();
+                } catch (final Throwable t) { // As per rule 2.13, `onComplete` is not allowed to throw exceptions, so we do what we can, and log this.
+                    (new IllegalStateException(subscriber + " violated the Reactive Streams rule 2.13 by throwing an exception from onComplete.", t)).printStackTrace(System.err);
                 }
             }
         }