Clarifies the signalling sequence in the spec and

 adds TCK verification to ensure signal ordering is proper.

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/IdentityProcessorVerification.java

@@ -286,6 +286,11 @@ public void optional_spec111_maySupportMultiSubscribe() throws Throwable {
     publisherVerification.optional_spec111_maySupportMultiSubscribe();
   }
 
+  @Override @Test
+  public void required_spec112_mustIssueOnSubscribeForNonNullSubscriber() throws Throwable {
+    publisherVerification.required_spec112_mustIssueOnSubscribeForNonNullSubscriber();
+  }
+
   @Override @Test
   public void required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe() throws Throwable {
     publisherVerification.required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe();

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());
           }
@@ -432,6 +441,7 @@ public void untested_spec108_possiblyCanceledSubscriptionShouldNotReceiveOnError
     notVerified(); // can we meaningfully test this?
   }
 
+  // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.9
   @Override @Test
   public void untested_spec109_subscribeShouldNotThrowNonFatalThrowable() throws Throwable {
     notVerified(); // can we meaningfully test this?
@@ -440,16 +450,16 @@ public void untested_spec109_subscribeShouldNotThrowNonFatalThrowable() throws T
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.9
   @Override @Test
   public void required_spec109_subscribeThrowNPEOnNullSubscriber() throws Throwable {
-      activePublisherTest(0, false, new PublisherTestRun<T>() {
-        @Override
-        public void run(Publisher<T> pub) throws Throwable {
-            try {
-                pub.subscribe(null);
-                env.flop(String.format("Publisher (%s) did not throw a NullPointerException when given a null Subscribe in subscribe", pub));
-            } catch (NullPointerException npe) {
-            }
-            env.verifyNoAsyncErrors();
+    activePublisherTest(0, false, new PublisherTestRun<T>() {
+      @Override
+      public void run(Publisher<T> pub) throws Throwable {
+        try {
+            pub.subscribe(null);
+            env.flop("Publisher did not throw a NullPointerException when given a null Subscribe in subscribe");
+        } catch (NullPointerException npe) {
         }
+        env.verifyNoAsyncErrors();
+      }
     });
   }
 
@@ -473,27 +483,66 @@ public void run(Publisher<T> pub) throws Throwable {
     });
   }
 
+  // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.12
+  @Override @Test
+  public void required_spec112_mustIssueOnSubscribeForNonNullSubscriber() throws Throwable {
+    activePublisherTest(0, false, new PublisherTestRun<T>() {
+      @Override
+      public void run(Publisher<T> pub) throws Throwable {
+        final Latch onSubscribeLatch = new Latch(env);
+        pub.subscribe(new Subscriber<T>() {
+          @Override
+          public void onError(Throwable cause) {
+            onSubscribeLatch.assertClosed("onSubscribe should be called prior to onError always");
+          }
+
+          @Override
+          public void onSubscribe(Subscription subs) {
+            onSubscribeLatch.assertOpen("Only one onSubscribe call expected");
+            onSubscribeLatch.close();
+          }
+
+          @Override
+          public void onNext(T elem) {
+            onSubscribeLatch.assertClosed("onSubscribe should be called prior to onNext always");
+          }
+
+          @Override
+          public void onComplete() {
+            onSubscribeLatch.assertClosed("onSubscribe should be called prior to onComplete always");
+          }
+        });
+        onSubscribeLatch.expectClose("Should have received onSubscribe");
+        env.verifyNoAsyncErrors();
+      }
+    });
+  }
+
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.12
   @Override @Test
   public void required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe() throws Throwable {
     whenHasErrorPublisherTest(new PublisherTestRun<T>() {
       @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/main/java/org/reactivestreams/tck/support/PublisherVerificationRules.java

@@ -24,6 +24,7 @@ public interface PublisherVerificationRules {
   void required_spec109_subscribeThrowNPEOnNullSubscriber() throws Throwable;
   void untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice() throws Throwable;
   void optional_spec111_maySupportMultiSubscribe() throws Throwable;
+  void required_spec112_mustIssueOnSubscribeForNonNullSubscriber() throws Throwable;
   void required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe() throws Throwable;
   void required_spec113_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingOneByOne() throws Throwable;
   void required_spec113_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingManyUpfront() throws Throwable;

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

@@ -250,6 +250,19 @@ public void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSigna
     }, "Unexpected element 0 received after stream completed");
   }
 
+  @Test
+  public void required_spec109_subscribeThrowNPEOnNullSubscriber_shouldFailIfDoesntThrowNPE() throws Throwable {
+    requireTestFailure(new ThrowingRunnable() {
+      @Override public void run() throws Throwable {
+        customPublisherVerification(new Publisher<Integer>() {
+          @Override public void subscribe(final Subscriber<? super Integer> s) {
+
+          }
+        }).required_spec109_subscribeThrowNPEOnNullSubscriber();
+      }
+    }, "Publisher did not throw a NullPointerException when given a null Subscribe in subscribe");
+  }
+
   @Test
   public void untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice_shouldFailBy_skippingSinceOptional() throws Throwable {
     requireTestFailure(new ThrowingRunnable() {
@@ -264,12 +277,52 @@ public void optional_spec111_maySupportMultiSubscribe_shouldFailBy_actuallyPass(
     noopPublisherVerification().optional_spec111_maySupportMultiSubscribe();
   }
 
+  @Test
+  public void required_spec112_mustIssueOnSubscribeForNonNullSubscriber_mustFailIfOnCompleteHappensFirst() throws Throwable {
+    requireTestFailure(new ThrowingRunnable() {
+      @Override public void run() throws Throwable {
+        customPublisherVerification(new Publisher<Integer>() {
+          @Override public void subscribe(Subscriber<? super Integer> s) {
+            s.onComplete();
+          }
+        }).required_spec112_mustIssueOnSubscribeForNonNullSubscriber();
+      }
+    }, "onSubscribe should be called prior to onComplete always");
+  }
+
+  @Test
+  public void required_spec112_mustIssueOnSubscribeForNonNullSubscriber_mustFailIfOnNextHappensFirst() throws Throwable {
+    requireTestFailure(new ThrowingRunnable() {
+      @Override public void run() throws Throwable {
+        customPublisherVerification(new Publisher<Integer>() {
+          @Override public void subscribe(Subscriber<? super Integer> s) {
+            s.onNext(1337);
+          }
+        }).required_spec112_mustIssueOnSubscribeForNonNullSubscriber();
+      }
+    }, "onSubscribe should be called prior to onNext always");
+  }
+
+  @Test
+  public void required_spec112_mustIssueOnSubscribeForNonNullSubscriber_mustFailIfOnErrorHappensFirst() throws Throwable {
+    requireTestFailure(new ThrowingRunnable() {
+      @Override public void run() throws Throwable {
+        customPublisherVerification(new Publisher<Integer>() {
+          @Override public void subscribe(Subscriber<? super Integer> s) {
+            s.onError(new TestException());
+          }
+        }).required_spec112_mustIssueOnSubscribeForNonNullSubscriber();
+      }
+    }, "onSubscribe should be called prior to onError always");
+  }
+
   @Test
   public void optional_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe_shouldFail() throws Throwable {
     requireTestFailure(new ThrowingRunnable() {
       @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 +333,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();