Modifies rules 1.09 and 2.13 to mandate java.lang.NullPointerException be thrown.

Updates the TCK, Spec and example implementations.

Author: viktorklang

README.md

@@ -66,6 +66,7 @@ final class SubscriptionImpl implements Subscription, Runnable {
     private Iterator<T> iterator; // This is our cursor into the data stream, which we will send to the `Subscriber`
 
     SubscriptionImpl(final Subscriber<? super T> subscriber) {
+      // As per rule 1.09, we need to throw a `java.lang.NullPointerException` if the `Subscriber` is `null`
       if (subscriber == null) throw null;
       this.subscriber = subscriber;
     }

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

@@ -143,15 +143,23 @@ private void handleOnError(final Throwable error) {
   // We implement the OnX methods on `Subscriber` to send Signals that we will process asycnhronously, but only one at a time
 
   @Override public final void onSubscribe(final Subscription s) {
+    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Subscription` is `null`
+    if (s == null) throw null;
     signal(new OnSubscribe(s));
   }
 
   @Override public final void onNext(final T element) {
+    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `element` is `null`
+    if (element == null) throw null;
+
     signal(new OnNext<T>(element));
   }
 
   @Override public final void onError(final Throwable t) {
-     signal(new OnError(t));
+    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Throwable` is `null`
+    if (t == null) throw null;
+
+    signal(new OnError(t));
   }
 
   @Override public final void onComplete() {

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

@@ -15,6 +15,9 @@ public abstract class SyncSubscriber<T> implements Subscriber<T> {
   private boolean done = false;
 
   @Override public void onSubscribe(final Subscription s) {
+    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Subscription` is `null`
+    if (s == null) throw null;
+
     if (subscription != null) { // If someone has made a mistake and added this Subscriber multiple times, let's handle it gracefully
       try {
         s.cancel(); // Cancel the additional subscription
@@ -38,6 +41,9 @@ public abstract class SyncSubscriber<T> implements Subscriber<T> {
   }
 
   @Override public void onNext(final T element) {
+    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `element` is `null`
+    if (element == null) throw null;
+
     if (!done) { // If we aren't already done
       try {
         if (foreach(element)) {
@@ -80,6 +86,8 @@ private void done() {
   protected abstract boolean foreach(final T element);
 
   @Override public void onError(final Throwable t) {
+    // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Throwable` is `null`
+    if (t == null) throw null;
      // Here we are not allowed to call any methods on the `Subscription` or the `Publisher`, as per rule 2.3
      // And anyway, the `Subscription` is considered to be cancelled if this method gets called, as per rule 2.4
   }

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

@@ -271,6 +271,11 @@ public void untested_spec109_subscribeShouldNotThrowNonFatalThrowable() throws T
     publisherVerification.untested_spec109_subscribeShouldNotThrowNonFatalThrowable();
   }
 
+  @Override @Test
+  public void required_spec109_subscribeThrowNPEOnNullSubscriber() throws Throwable {
+    publisherVerification.required_spec109_subscribeThrowNPEOnNullSubscriber();
+  }
+
   @Override @Test
   public void untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice() throws Throwable {
     publisherVerification.untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice();
@@ -558,6 +563,11 @@ public void untested_spec213_failingOnSignalInvocation() throws Exception {
     subscriberVerification.untested_spec213_failingOnSignalInvocation();
   }
 
+  @Override @Test
+  public void required_spec213_mustThrowNullPointerExceptionWhenParametersAreNull() throws Throwable {
+    subscriberVerification.required_spec213_mustThrowNullPointerExceptionWhenParametersAreNull();
+  }
+
   @Override @Test
   public void untested_spec301_mustNotBeCalledOutsideSubscriberContext() throws Exception {
     subscriberVerification.untested_spec301_mustNotBeCalledOutsideSubscriberContext();

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

@@ -432,10 +432,24 @@ 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(); // cannot be meaningfully tested, or can it?
+    notVerified(); // can we meaningfully test this?
+  }
+
+  // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.9
+  @Override @Test
+  public void required_spec109_subscribeThrowNPEOnNullSubscriber() throws Throwable {
+    optionalActivePublisherTest(1, true, 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();
+      }
+    });
   }
 
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.10

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

@@ -14,10 +14,14 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.lang.NullPointerException;
+import java.lang.Override;
+import java.lang.RuntimeException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
 import static org.reactivestreams.tck.SubscriberWhiteboxVerification.BlackboxSubscriberProxy;
+import static org.testng.Assert.assertTrue;
 
 /**
  * Provides tests for verifying {@link org.reactivestreams.Subscriber} and {@link org.reactivestreams.Subscription}
@@ -310,6 +314,57 @@ public void untested_spec213_blackbox_failingOnSignalInvocation() throws Excepti
     notVerified(); // cannot be meaningfully tested, or can it?
   }
 
+  // Verifies rule: https://github.com/reactive-streams/reactive-streams#2.13
+  @Override @Test
+  public void required_spec213_blackbox_mustThrowNullPointerExceptionWhenParametersAreNull() throws Throwable {
+    blackboxSubscriberWithoutSetupTest(new BlackboxTestStageTestRun() {
+      @Override
+      public void run(BlackboxTestStage stage) throws Throwable {
+        final Subscription subscription = new Subscription() {
+          @Override public void request(final long elements) {}
+          @Override public void cancel() {}
+        };
+
+        {
+          final Subscriber<T> sub = createSubscriber();
+          boolean gotNPE = false;
+          try {
+            sub.onSubscribe(null);
+          } catch(final NullPointerException expected) {
+            gotNPE = true;
+          }
+          assertTrue(gotNPE, "onSubscribe(null) did not throw NullPointerException");
+        }
+
+        {
+          final Subscriber<T> sub = createSubscriber();
+          boolean gotNPE = false;
+          sub.onSubscribe(subscription);
+          try {
+            sub.onNext(null);
+          } catch(final NullPointerException expected) {
+            gotNPE = true;
+          }
+          assertTrue(gotNPE, "onNext(null) did not throw NullPointerException");
+        }
+
+        {
+          final Subscriber<T> sub = createSubscriber();
+          boolean gotNPE = false;
+          sub.onSubscribe(subscription);
+          try {
+            sub.onError(null);
+          } catch(final NullPointerException expected) {
+            gotNPE = true;
+          }
+          assertTrue(gotNPE, "onError(null) did not throw NullPointerException");
+        }
+
+        env.verifyNoAsyncErrors();
+      }
+    });
+  }
+
   ////////////////////// SUBSCRIPTION SPEC RULE VERIFICATION //////////////////
 
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#3.1

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

@@ -343,6 +343,59 @@ public void untested_spec213_failingOnSignalInvocation() throws Exception {
     notVerified(); // cannot be meaningfully tested, or can it?
   }
 
+  // Verifies rule: https://github.com/reactive-streams/reactive-streams#2.13
+  @Override @Test
+  public void required_spec213_mustThrowNullPointerExceptionWhenParametersAreNull() throws Throwable {
+    subscriberTest(new TestStageTestRun() {
+      @Override
+      public void run(WhiteboxTestStage stage) throws Throwable {
+
+        final Subscription subscription = new Subscription() {
+          @Override public void request(final long elements) {}
+          @Override public void cancel() {}
+        };
+
+        {
+          final Subscriber<T> sub = createSubscriber(stage.probe());
+          boolean gotNPE = false;
+          try {
+            sub.onSubscribe(null);
+          } catch(final NullPointerException expected) {
+            gotNPE = true;
+          }
+          assertTrue(gotNPE, "onSubscribe(null) did not throw NullPointerException");
+        }
+
+        {
+          final Subscriber<T> sub = createSubscriber(stage.probe());
+          boolean gotNPE = false;
+          sub.onSubscribe(subscription);
+          try {
+            sub.onNext(null);
+          } catch(final NullPointerException expected) {
+            gotNPE = true;
+          }
+          assertTrue(gotNPE, "onNext(null) did not throw NullPointerException");
+        }
+
+        {
+          final Subscriber<T> sub = createSubscriber(stage.probe());
+          boolean gotNPE = false;
+          sub.onSubscribe(subscription);
+          try {
+            sub.onError(null);
+          } catch(final NullPointerException expected) {
+            gotNPE = true;
+          }
+          assertTrue(gotNPE, "onError(null) did not throw NullPointerException");
+        }
+
+        env.verifyNoAsyncErrors();
+      }
+    });
+  }
+
+
   ////////////////////// SUBSCRIPTION SPEC RULE VERIFICATION //////////////////
 
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#3.1

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

@@ -21,6 +21,7 @@ public interface PublisherVerificationRules {
   void untested_spec107_mustNotEmitFurtherSignalsOnceOnErrorHasBeenSignalled() throws Throwable;
   void untested_spec108_possiblyCanceledSubscriptionShouldNotReceiveOnErrorOrOnCompleteSignals() throws Throwable;
   void untested_spec109_subscribeShouldNotThrowNonFatalThrowable() throws Throwable;
+  void required_spec109_subscribeThrowNPEOnNullSubscriber() throws Throwable;
   void untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice() throws Throwable;
   void optional_spec111_maySupportMultiSubscribe() throws Throwable;
   void required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe() throws Throwable;

tck/src/main/java/org/reactivestreams/tck/support/PublisherVerificationRules.java

@@ -20,6 +20,7 @@ public interface SubscriberBlackboxVerificationRules {
   void untested_spec211_blackbox_mustMakeSureThatAllCallsOnItsMethodsHappenBeforeTheProcessingOfTheRespectiveEvents() throws Exception;
   void untested_spec212_blackbox_mustNotCallOnSubscribeMoreThanOnceBasedOnObjectEquality() throws Throwable;
   void untested_spec213_blackbox_failingOnSignalInvocation() throws Exception;
+  void required_spec213_blackbox_mustThrowNullPointerExceptionWhenParametersAreNull() throws Throwable;
   void untested_spec301_blackbox_mustNotBeCalledOutsideSubscriberContext() throws Exception;
   void required_spec308_blackbox_requestMustRegisterGivenNumberElementsToBeProduced() throws Throwable;
   void untested_spec310_blackbox_requestMaySynchronouslyCallOnNextOnSubscriber() throws Exception;

tck/src/main/java/org/reactivestreams/tck/support/SubscriberBlackboxVerificationRules.java

@@ -22,6 +22,7 @@ public interface SubscriberWhiteboxVerificationRules {
   void untested_spec211_mustMakeSureThatAllCallsOnItsMethodsHappenBeforeTheProcessingOfTheRespectiveEvents() throws Exception;
   void untested_spec212_mustNotCallOnSubscribeMoreThanOnceBasedOnObjectEquality_specViolation() throws Throwable;
   void untested_spec213_failingOnSignalInvocation() throws Exception;
+  void required_spec213_mustThrowNullPointerExceptionWhenParametersAreNull() throws Throwable;
   void untested_spec301_mustNotBeCalledOutsideSubscriberContext() throws Exception;
   void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced() throws Throwable;
   void untested_spec310_requestMaySynchronouslyCallOnNextOnSubscriber() throws Exception;