Merge pull request #373 from ktoso/wip-205-complete-once-done

+tck #362 complete subscriber under test once done in 205

Author: viktorklang

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

@@ -227,6 +227,7 @@ public String toString() {
 
       secondSubscriptionCancelled.expectClose("Expected SecondSubscription given to subscriber to be cancelled, but `Subscription.cancel()` was not called.");
       env.verifyNoAsyncErrorsNoDelay();
+      sendCompletion(); // we're done, complete the subscriber under test
     }};
   }
 

Diff for: tck/src/test/java/org/reactivestreams/tck/SubscriberBlackboxVerificationTest.java

@@ -18,8 +18,10 @@
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 /**
 * Validates that the TCK's {@link org.reactivestreams.tck.SubscriberBlackboxVerification} fails with nice human readable errors.
@@ -111,6 +113,34 @@ public void required_spec205_blackbox_mustCallSubscriptionCancelIfItAlreadyHasAn
     }, "illegally called `subscription.request(1)");
   }
 
+  @Test
+  public void required_spec205_blackbox_mustCallSubscriptionCancelIfItAlreadyHasAnSubscriptionAndReceivesAnotherOnSubscribeSignal_shouldGetCompletion() throws Throwable {
+    final CountDownLatch completion = new CountDownLatch(1);
+      
+    customSubscriberVerification(new KeepSubscriptionSubscriber() {
+      volatile Subscription sub;
+
+      @Override
+      public void onSubscribe(Subscription s) {
+        super.onSubscribe(s);
+        if (sub != null) {
+          sub = s;
+          s.request(1);
+        } else {
+          // the second one we cancel
+          s.cancel();
+        }
+      }
+
+      @Override
+      public void onComplete() {
+        completion.countDown();
+      }
+    }).required_spec205_blackbox_mustCallSubscriptionCancelIfItAlreadyHasAnSubscriptionAndReceivesAnotherOnSubscribeSignal();
+    
+    completion.await(1, TimeUnit.SECONDS);
+  }
+
   @Test
   public void required_spec209_blackbox_mustBePreparedToReceiveAnOnCompleteSignalWithPrecedingRequestCall_shouldFail() throws Throwable {
     requireTestFailure(new ThrowingRunnable() {