Attempt to clarify the signalling sequence in the spec

Author: viktorklang

README.md

@@ -57,9 +57,12 @@ A *Publisher* is a provider of a potentially unbounded number of sequenced eleme
 In response to a call to `Publisher.subscribe(Subscriber)` the possible invocation sequences for methods on the `Subscriber` are given by the following protocol:
 
 ```
-onError | (onSubscribe onNext* (onError | onComplete)?)
+onSubscribe onNext* (onError | onComplete)?
 ```
 
+This means that `onSubscribe` is always signalled,
+followed by a possibly unbounded number of `onNext` signals (as requested by `Subscriber`) followed by an `onError` signal if there is a failure, or an `onComplete` signal when no more elements are available—all as long as the `Subscription` is not cancelled.
+
 #### NOTES
 
 - The specifications below use binding words in capital letters from https://www.ietf.org/rfc/rfc2119.txt
@@ -87,10 +90,10 @@ public interface Publisher<T> {
 | <a name="1.6">6</a>       | If a `Publisher` signals either `onError` or `onComplete` on a `Subscriber`, that `Subscriber`’s `Subscription` MUST be considered cancelled. |
 | <a name="1.7">7</a>       | Once a terminal state has been signaled (`onError`, `onComplete`) it is REQUIRED that no further signals occur. |
 | <a name="1.8">8</a>       | If a `Subscription` is cancelled its `Subscriber` MUST eventually stop being signaled. |
-| <a name="1.9">9</a>       | Calling `Publisher.subscribe` MUST return normally except when the provided `Subscriber` is `null` in which case it MUST throw a `java.lang.NullPointerException` to the caller, for all other situations the only legal way to signal failure (or reject a `Subscriber`) is via the `onError` method. |
+| <a name="1.9">9</a>       | `Publisher.subscribe` MUST signal `onSubscribe` on the provided `Subscriber` prior to any other signals to that `Subscriber` and MUST return normally, except when the provided `Subscriber` is `null` in which case it MUST throw a `java.lang.NullPointerException` to the caller, for all other situations the only legal way to signal failure (or reject the `Subscriber`) is by calling `onError` (after calling `onSubscribe`). |
 | <a name="1.10">10</a>     | `Publisher.subscribe` MAY be called as many times as wanted but MUST be with a different `Subscriber` each time [see [2.12](#2.12)]. |
 | <a name="1.11">11</a>     | A `Publisher` MAY support multi-subscribe and choose whether each `Subscription` is unicast or multicast. |
-| <a name="1.12">12</a>     | A `Publisher` MAY reject calls to its `subscribe` method if it is unable or unwilling to serve them [[1](#footnote-1-1)]. If rejecting it MUST do this by calling `onError` on the `Subscriber` passed to `Publisher.subscribe` instead of calling `onSubscribe`. |
+| <a name="1.12">12</a>     | A `Publisher` MAY reject calls to its `subscribe` method if it is unable or unwilling to serve them [[1](#footnote-1-1)]. If rejecting it MUST do this by calling `onError` on the `Subscriber` passed to `Publisher.subscribe` after calling `onSubscribe`. |
 | <a name="1.13">13</a>     | A `Publisher` MUST produce the same elements, starting with the oldest element still available, in the same sequence for all its subscribers and MAY produce the stream elements at (temporarily) differing rates to different subscribers. |
 
 [<a name="footnote-1-1">1</a>] :  A stateful Publisher can be overwhelmed, bounded by a finite number of underlying resources, exhausted, shut-down or in a failed state.

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

@@ -106,6 +106,10 @@ private void doSubscribe() {
         if (iterator == null)
           iterator = Collections.<T>emptyList().iterator(); // So we can assume that `iterator` is never null
       } catch(final Throwable t) {
+        subscriber.onSubscribe(new Subscription() { // We need to make sure we signal onSubscribe before onError, obeying rule 1.12
+          @Override public void cancel() {}
+          @Override public void request(long n) {}
+        });
         terminateDueTo(t); // Here we send onError, obeying rule 1.12
       }
 

examples/src/test/java/org/reactivestreams/example/unicast/IterablePublisherTest.java

@@ -1,5 +1,6 @@
 package org.reactivestreams.example.unicast;
 
+import java.lang.Override;
 import java.util.Collections;
 import java.util.Iterator;
 import org.reactivestreams.Publisher;
@@ -30,7 +31,11 @@ public IterablePublisherTest() {
   }
 
   @Override public Publisher<Integer> createErrorStatePublisher() {
-    return null;
+    return new AsyncIterablePublisher<Integer>(new Iterable<Integer>() {
+      @Override public Iterator<Integer> iterator() {
+        throw new RuntimeException("Error state signal!");
+      }
+    }, e);
   }
 
   @Override public long maxElementsFromPublisher() {

examples/src/test/java/org/reactivestreams/example/unicast/UnboundedIntegerIncrementPublisherTest.java

@@ -8,6 +8,7 @@
 import org.testng.annotations.AfterClass;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ExecutorService;
+import java.util.Iterator;
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class UnboundedIntegerIncrementPublisherTest extends PublisherVerification<Integer> {
@@ -25,7 +26,11 @@ public UnboundedIntegerIncrementPublisherTest() {
   }
 
   @Override public Publisher<Integer> createErrorStatePublisher() {
-    return null;
+    return new AsyncIterablePublisher<Integer>(new Iterable<Integer>() {
+      @Override public Iterator<Integer> iterator() {
+        throw new RuntimeException("Error state signal!");
+      }
+    }, e);
   }
 
   @Override public long maxElementsFromPublisher() {

tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java

@@ -14,6 +14,7 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.lang.Override;
 import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
 import java.util.ArrayList;
@@ -344,16 +345,24 @@ public void optional_spec104_mustSignalOnErrorWhenFails() throws Throwable {
       whenHasErrorPublisherTest(new PublisherTestRun<T>() {
         @Override
         public void run(final Publisher<T> pub) throws InterruptedException {
-          final Latch latch = new Latch(env);
+          final Latch onErrorlatch = new Latch(env);
+          final Latch onSubscribeLatch = new Latch(env);
           pub.subscribe(new TestEnvironment.TestSubscriber<T>(env) {
+            @Override
+            public void onSubscribe(Subscription subs) {
+              onSubscribeLatch.assertOpen("Only one onSubscribe call expected");
+              onSubscribeLatch.close();
+            }
             @Override
             public void onError(Throwable cause) {
-              latch.assertOpen(String.format("Error-state Publisher %s called `onError` twice on new Subscriber", pub));
-              latch.close();
+              onSubscribeLatch.assertClosed("onSubscribe should be called prior to onError always");
+              onErrorlatch.assertOpen(String.format("Error-state Publisher %s called `onError` twice on new Subscriber", pub));
+              onErrorlatch.close();
             }
           });
 
-          latch.expectClose(String.format("Error-state Publisher %s did not call `onError` on new Subscriber", pub));
+          onSubscribeLatch.expectClose("Should have received onSubscribe");
+          onErrorlatch.expectClose(String.format("Error-state Publisher %s did not call `onError` on new Subscriber", pub));
 
           env.verifyNoAsyncErrors(env.defaultTimeoutMillis());
           }
@@ -480,20 +489,24 @@ public void required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwil
       @Override
       public void run(Publisher<T> pub) throws Throwable {
         final Latch onErrorLatch = new Latch(env);
+        final Latch onSubscribeLatch = new Latch(env);
         ManualSubscriberWithSubscriptionSupport<T> sub = new ManualSubscriberWithSubscriptionSupport<T>(env) {
           @Override
           public void onError(Throwable cause) {
+            onSubscribeLatch.assertClosed("onSubscribe should be called prior to onError always");
             onErrorLatch.assertOpen("Only one onError call expected");
             onErrorLatch.close();
           }
 
           @Override
           public void onSubscribe(Subscription subs) {
-            env.flop("onSubscribe should not be called if Publisher is unable to subscribe a Subscriber");
+            onSubscribeLatch.assertOpen("Only one onSubscribe call expected");
+            onSubscribeLatch.close();
           }
         };
         pub.subscribe(sub);
-        onErrorLatch.assertClosed("Should have received onError");
+        onSubscribeLatch.expectClose("Should have received onSubscribe");
+        onErrorLatch.expectClose("Should have received onError");
 
         env.verifyNoAsyncErrors();
         }

tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java

@@ -270,6 +270,7 @@ public void optional_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwil
       @Override public void run() throws Throwable {
         customPublisherVerification(SKIP, new Publisher<Integer>() {
           @Override public void subscribe(Subscriber<? super Integer> s) {
+            s.onSubscribe(new NoopSubscription());
           }
         }).required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe();
       }
@@ -280,6 +281,7 @@ public void optional_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwil
   public void optional_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe_actuallyPass() throws Throwable {
     customPublisherVerification(SKIP, new Publisher<Integer>() {
       @Override public void subscribe(Subscriber<? super Integer> s) {
+        s.onSubscribe(new NoopSubscription());
         s.onError(new RuntimeException("Sorry, I'm busy now. Call me later."));
       }
     }).required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe();