Merge pull request #376 from egetman/feature/tck-optional-111-reference-drop-fix

optional_spec111 test addition.

Author: viktorklang

CopyrightWaivers.txt

@@ -36,3 +36,4 @@ JakeWharton    | Jake Wharton, [email protected]
 anthonyvdotbe  | Anthony Vanelverdinghe, [email protected]
 seratch        | Kazuhiro Sera, [email protected], SmartNews, Inc.
 akarnokd       | David Karnok, [email protected]
+egetman        | Evgeniy Getman, [email protected]

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

@@ -311,6 +311,11 @@ public void optional_spec111_maySupportMultiSubscribe() throws Throwable {
     publisherVerification.optional_spec111_maySupportMultiSubscribe();
   }
 
+  @Override @Test
+  public void optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals() throws Throwable {
+    publisherVerification.optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals();
+  }
+
   @Override @Test
   public void optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingOneByOne() throws Throwable {
     publisherVerification.optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingOneByOne();

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

@@ -572,6 +572,33 @@ public void run(Publisher<T> pub) throws Throwable {
     });
   }
 
+  @Override @Test
+  public void optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals() throws Throwable {
+    optionalActivePublisherTest(1, false, new PublisherTestRun<T>() {
+      @Override
+      public void run(Publisher<T> pub) throws Throwable {
+        ManualSubscriber<T> sub1 = env.newManualSubscriber(pub);
+        ManualSubscriber<T> sub2 = env.newManualSubscriber(pub);
+        // Since we're testing the case when the Publisher DOES support the optional multi-subscribers scenario,
+        // and decides if it handles them uni-cast or multi-cast, we don't know which subscriber will receive an
+        // onNext (and optional onComplete) signal(s) and which just onComplete signal.
+        // Plus, even if subscription assumed to be unicast, it's implementation choice, which one will be signalled
+        // with onNext.
+        sub1.requestNextElementOrEndOfStream();
+        sub2.requestNextElementOrEndOfStream();
+        try {
+            env.verifyNoAsyncErrors();
+        } finally {
+            try {
+                sub1.cancel();
+            } finally {
+                sub2.cancel();
+            }
+        }
+      }
+    });
+  }
+
   @Override @Test
   public void optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingOneByOne() throws Throwable {
     optionalActivePublisherTest(5, true, new PublisherTestRun<T>() { // This test is skipped if the publisher is unbounded (never sends onComplete)

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

@@ -390,6 +390,10 @@ public T requestNextElement(long timeoutMillis, String errorMsg) throws Interrup
       return nextElement(timeoutMillis, errorMsg);
     }
 
+    public Optional<T> requestNextElementOrEndOfStream() throws InterruptedException {
+      return requestNextElementOrEndOfStream(env.defaultTimeoutMillis(), "Did not receive expected stream completion");
+    }
+
     public Optional<T> requestNextElementOrEndOfStream(String errorMsg) throws InterruptedException {
       return requestNextElementOrEndOfStream(env.defaultTimeoutMillis(), errorMsg);
     }

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

@@ -285,7 +285,8 @@ public interface PublisherVerificationRules {
    */
   void untested_spec110_rejectASubscriptionRequestIfTheSameSubscriberSubscribesTwice() throws Throwable;
   /**
-   * Ask for a single-element {@code Publisher} and subscribes to it twice, without consuming with either {@code Subscriber} instance
+   * Asks for a single-element {@code Publisher} and subscribes to it twice, without consuming with either
+   * {@code Subscriber} instance
    * (i.e., no requests are issued).
    * <p>
    * <b>Verifies rule:</b> <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.11'>1.11</a>
@@ -296,6 +297,19 @@ public interface PublisherVerificationRules {
    * means will indicate a skipped test.
    */
   void optional_spec111_maySupportMultiSubscribe() throws Throwable;
+  /**
+   * Asks for a single-element {@code Publisher} and subscribes to it twice.
+   * Each {@code Subscriber} requests for 1 element and checks if onNext or onComplete signals was received.
+   * <p>
+   * <b>Verifies rule:</b> <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.11'>1.11</a>,
+   * and depends on valid implementation of rule <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.5'>1.5</a>
+   * in order to verify this.
+   * <p>
+   * The test is not executed if {@link org.reactivestreams.tck.PublisherVerification#maxElementsFromPublisher()} is less than 1.
+   * <p>
+   * Any exception thrown through non-regular means will indicate a skipped test.
+   */
+  void optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals() throws Throwable;
   /**
    * Asks for a short {@code Publisher} (length 5), subscribes 3 {@code Subscriber}s to it, requests with different
    * patterns and checks if all 3 received the same events in the same order.
@@ -310,7 +324,7 @@ public interface PublisherVerificationRules {
    * when they subscribe and how they request elements. I.e., a "live" {@code Publisher} emitting the current time would not pass this test.
    * <p>
    * Note that this test is optional and may appear skipped even if the behavior should be actually supported by the {@code Publisher},
-   * see the skip message for an indication of this. 
+   * see the skip message for an indication of this.
    * <p>
    * If this test fails, the following could be checked within the {@code Publisher} implementation:
    * <ul>

tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java

@@ -19,13 +19,16 @@
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import java.util.Collection;
 import java.util.Random;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
 * Validates that the TCK's {@link org.reactivestreams.tck.PublisherVerification} fails with nice human readable errors.
@@ -364,6 +367,21 @@ public void optional_spec111_maySupportMultiSubscribe_shouldFailBy_actuallyPass(
     noopPublisherVerification().optional_spec111_maySupportMultiSubscribe();
   }
 
+  @Test
+  public void optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals_beSkippedWhenMultipleSubscribersNotSupported() throws Throwable {
+    requireTestSkip(new ThrowingRunnable() {
+      @Override
+      public void run() throws Throwable {
+        multiSubscribersPublisherVerification(true).optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals();
+      }
+    }, "Unexpected additional subscriber");
+  }
+
+  @Test
+  public void optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals_shouldPass() throws Throwable {
+    multiSubscribersPublisherVerification(false).optional_spec111_registeredSubscribersMustReceiveOnNextOrOnCompleteSignals();
+  }
+
   @Test
   public void optional_spec111_multicast_mustProduceTheSameElementsInTheSameSequenceToAllOfItsSubscribersWhenRequestingManyUpfront_shouldFailBy_expectingOnError() throws Throwable {
     requireTestFailure(new ThrowingRunnable() {
@@ -716,6 +734,73 @@ final PublisherVerification<Integer> customPublisherVerification(final Publisher
     };
   }
 
+  /**
+   * Verification using a Publisher that supports multiple subscribers
+   * @param shouldBlowUp if true {@link RuntimeException} will be thrown during second subscription.
+   */
+  final PublisherVerification<Integer> multiSubscribersPublisherVerification(final boolean shouldBlowUp) {
+    return new PublisherVerification<Integer>(newTestEnvironment()) {
+
+      @Override
+      public Publisher<Integer> createPublisher(final long elements) {
+        return new Publisher<Integer>() {
+
+          private final Collection<CancelableSubscription> subscriptions = new CopyOnWriteArrayList<CancelableSubscription>();
+          private final AtomicLong source = new AtomicLong(elements);
+
+          @Override
+          public void subscribe(Subscriber<? super Integer> s) {
+            // onSubscribe first
+            CancelableSubscription subscription = new CancelableSubscription(s);
+            s.onSubscribe(subscription);
+            if (shouldBlowUp && !subscriptions.isEmpty()) {
+              s.onError(new RuntimeException("Unexpected additional subscriber"));
+            } else {
+              subscriptions.add(subscription);
+            }
+          }
+
+          class CancelableSubscription implements Subscription {
+
+            final AtomicBoolean canceled = new AtomicBoolean();
+            Subscriber<? super Integer> subscriber;
+
+            CancelableSubscription(Subscriber<? super Integer> subscriber) {
+              this.subscriber = subscriber;
+            }
+
+            @Override
+            public void request(long n) {
+              if (!canceled.get()) {
+                for (long i = 0; i < n; i++) {
+                  if (source.getAndDecrement() < 0) {
+                    canceled.set(true);
+                    subscriber.onComplete();
+                  } else {
+                    subscriber.onNext((int) i);
+                  }
+                }
+              }
+            }
+
+            @Override
+            public void cancel() {
+              canceled.set(true);
+              subscriber = null;
+              subscriptions.remove(this);
+            }
+          }
+
+        };
+      }
+
+      @Override
+      public Publisher<Integer> createFailedPublisher() {
+        return SKIP;
+      }
+    };
+  }
+
   /**
    * Verification using a Publisher that publishes elements even with no demand available
    */