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

@@ -437,8 +437,30 @@ 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();
+        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) {
+            stillBeingSignalled = true;
+          }
+        } while (stillBeingSignalled);
 
         stage.verifyNoAsyncErrors();
       }
@@ -518,7 +540,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 +568,10 @@ public SubscriberPuppet puppet() {
       return probe.puppet();
     }
 
+    public WhiteboxSubscriberProbe<T> probe() {
+      return probe;
+    }
+
     public Publisher<T> createHelperPublisher(long elements) {
       return SubscriberWhiteboxVerification.this.createHelperPublisher(elements);
     }