Merge pull request #372 from ktoso/wip-complete-in-201

=tck #362 signal onComplete in 201 blackbox verification

Author: viktorklang

Diff for: tck/src/main/java/org/reactivestreams/tck/SubscriberBlackboxVerification.java

@@ -95,11 +95,25 @@ public void required_spec201_blackbox_mustSignalDemandViaSubscriptionRequest() t
       @Override
       public void run(BlackboxTestStage stage) throws InterruptedException {
         triggerRequest(stage.subProxy().sub());
-        final long n = stage.expectRequest();// assuming subscriber wants to consume elements...
+        final long requested = stage.expectRequest();// assuming subscriber wants to consume elements...
+        final long signalsToEmit = Math.min(requested, 512); // protecting against Subscriber which sends ridiculous large demand
 
         // should cope with up to requested number of elements
-        for (int i = 0; i < n; i++)
+        for (int i = 0; i < signalsToEmit && sampleIsCancelled(stage, i, 10); i++)
           stage.signalNext();
+        
+        // we complete after `signalsToEmit` (which can be less than `requested`), 
+        // which is legal under https://github.com/reactive-streams/reactive-streams-jvm#1.2
+        stage.sendCompletion();
+      }
+
+      /** 
+       * In order to allow some "skid" and not check state on each iteration, 
+       * only check {@code stage.isCancelled} every {@code checkInterval}'th iteration.
+       */
+      private boolean sampleIsCancelled(BlackboxTestStage stage, int i, int checkInterval) throws InterruptedException {
+        if (i % checkInterval == 0) return stage.isCancelled();
+        else return false;
       }
     });
   }

Diff for: tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java

@@ -775,6 +775,10 @@ public void expectCancelling() throws InterruptedException {
     public void expectCancelling(long timeoutMillis) throws InterruptedException {
       cancelled.expectClose(timeoutMillis, "Did not receive expected cancelling of upstream subscription");
     }
+    
+    public boolean isCancelled() throws InterruptedException {
+      return cancelled.isClosed();
+    }
   }
 
   /**