Rule 3.12 verification in subscriber whitebox now really checks it

Resolves reactive-streams#115

Author: ktoso

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

@@ -386,9 +386,6 @@ public void onSubscribe(final Subscription subscription) {
             if (subs.isCompleted()) subscription.cancel(); // the Probe must also pass subscriber verification
 
             probe.registerOnSubscribe(new SubscriberWhiteboxVerification.SubscriberPuppet() {
-              public void triggerShutdown() {
-                subscription.cancel();
-              }
 
               @Override
               public void triggerRequest(long elements) {

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

@@ -52,6 +52,17 @@ protected SubscriberWhiteboxVerification(TestEnvironment env) {
    */
   public abstract Publisher<T> createHelperPublisher(long elements);
 
+  /**
+   * Used to break possibly infinite wait-loops.
+   * Some Rules use the "eventually stop signalling" wording, which requires the test to spin accepting {@code onNext}
+   * signals until no more are signalled. In these tests, this value will be used as upper bound on the number of spin iterations.
+   *
+   * Override this method in case your implementation synchronously signals very large batches before reacting to cancellation (for example).
+   */
+  public long maxOnNextSignalsInTest() {
+    return 100;
+  }
+
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
 
   @BeforeMethod
@@ -437,8 +448,33 @@ public void spec312_cancelMustRequestThePublisherToEventuallyStopSignaling() thr
     subscriberTest(new TestStageTestRun() {
       @Override
       public void run(WhiteboxTestStage stage) throws InterruptedException {
+        final int requestedElements = 5;
+
+        stage.puppet().triggerRequest(requestedElements);
+        stage.expectRequest();
+        stage.signalNext();
+
+        // cancelling is NOT REQUIRED to stop the stream immediatly, example: there may be signals in flight
         stage.puppet().signalCancel();
-        stage.expectCancelling();
+
+        stage.probe().expectNext();
+        long onNextSignals = 1;
+
+        boolean stillBeingSignalled;
+        do {
+          // depends on Rule 3.6 https://github.com/reactive-streams/reactive-streams#3.12
+          // requesting from a cancelled subscription MUST be a NOPs
+          stage.puppet().triggerRequest(1);
+          stage.signalNext(); // signalling will stop, since Subscription cancelled
+
+          try {
+            stage.probe().expectNone();
+            stillBeingSignalled = false;
+          } catch (Throwable th) {
+            onNextSignals += 1;
+            stillBeingSignalled = true;
+          }
+        } while (stillBeingSignalled && onNextSignals < maxOnNextSignalsInTest());
 
         stage.verifyNoAsyncErrors();
       }
@@ -518,7 +554,7 @@ public void subscriberTestWithoutSetup(TestStageTestRun body) throws Throwable {
 
   public class WhiteboxTestStage extends ManualPublisher<T> {
     public Publisher<T> pub;
-    public ManualSubscriber<T> tees; // gives us access to an infinite stream of T values
+    public ManualSubscriber<T> tees; // gives us access to a stream T values
     public WhiteboxSubscriberProbe<T> probe;
 
     public T lastT = null;
@@ -546,6 +582,10 @@ public SubscriberPuppet puppet() {
       return probe.puppet();
     }
 
+    public WhiteboxSubscriberProbe<T> probe() {
+      return probe;
+    }
+
     public Publisher<T> createHelperPublisher(long elements) {
       return SubscriberWhiteboxVerification.this.createHelperPublisher(elements);
     }