Allow skipping tests which require more subscribers

if your impl is not able to support multiple subscribers

Author: ktoso

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

@@ -33,6 +33,16 @@ private Annotations() {}
   static @interface Required {
   }
 
+  /**
+   * Used to mark how many subscribers will be used on the publisher under test.
+   * If an implementation declares it does not support enough subscribers as needed by such test, it will be SKIPPed instead of FAILed.
+   */
+  @Target(ElementType.METHOD)
+  @Retention(RetentionPolicy.SOURCE)
+  static @interface Subscribers {
+    long value() default 1;
+  }
+
   /**
    * Used to mark tests which may be skipped / not implemented by certain implementations.
    * These tests can be skipped by returning null from the given factory method.

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

@@ -4,10 +4,12 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.reactivestreams.tck.Annotations.Subscribers;
 import org.reactivestreams.tck.TestEnvironment.ManualPublisher;
 import org.reactivestreams.tck.TestEnvironment.ManualSubscriber;
 import org.reactivestreams.tck.TestEnvironment.ManualSubscriberWithSubscriptionSupport;
 import org.reactivestreams.tck.TestEnvironment.Promise;
+import org.reactivestreams.tck.support.Function;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
@@ -130,17 +132,28 @@ public long maxElementsFromPublisher() {
   }
 
   /**
-     * In order to verify rule 3.3 of the reactive streams spec, this number will be used to check if a
-     * {@code Subscription} actually solves the "unbounded recursion" problem by not allowing the number of
-     * recursive calls to exceed the number returned by this method.
-     *
-     * @see <a href="https://github.com/reactive-streams/reactive-streams#3.3">reactive streams spec, rule 3.3</a>
-     * @see PublisherVerification#spec303_mustNotAllowUnboundedRecursion()
-     */
+   * In order to verify rule 3.3 of the reactive streams spec, this number will be used to check if a
+   * {@code Subscription} actually solves the "unbounded recursion" problem by not allowing the number of
+   * recursive calls to exceed the number returned by this method.
+   *
+   * @see <a href="https://github.com/reactive-streams/reactive-streams#3.3">reactive streams spec, rule 3.3</a>
+   * @see PublisherVerification#spec303_mustNotAllowUnboundedRecursion()
+   */
   public long boundedDepthOfOnNextAndRequestRecursion() {
     return 1;
   }
 
+  /**
+   * Describes the tested implementation in terms of how many subscribers they can support.
+   * Some tests require the {@code Publisher} under test to support multiple Subscribers,
+   * yet the spec does not require all publishers to be able to do so, thus – if an implementation
+   * supports only a limited number of subscribers (e.g. only 1 subscriber, also known as "no fanout")
+   * you MUST return that number from this method by overriding it.
+   */
+  public long maxSupportedSubscribers() {
+      return Long.MAX_VALUE;
+  }
+
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
 
   @BeforeMethod
@@ -310,31 +323,37 @@ public void spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue()
 
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.4
   // for multiple subscribers
-  @Test
-  public void spec104_mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError() throws Exception {
-    new TestSetup(env, processorBufferSize) {{
-      final ManualSubscriberWithErrorCollection<T> sub1 = new ManualSubscriberWithErrorCollection<T>(env);
-      env.subscribe(processor, sub1);
+  @Test @Subscribers(2)
+  public void spec104_mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError() throws Throwable {
+    optionalMultipleSubscribersTest(2, new Function<Long,TestSetup>() {
+      @Override
+      public TestSetup apply(Long aLong) throws Throwable {
+        return new TestSetup(env, processorBufferSize) {{
+          final ManualSubscriberWithErrorCollection<T> sub1 = new ManualSubscriberWithErrorCollection<T>(env);
+          env.subscribe(processor, sub1);
 
-      final ManualSubscriberWithErrorCollection<T> sub2 = new ManualSubscriberWithErrorCollection<T>(env);
-      env.subscribe(processor, sub2);
+          final ManualSubscriberWithErrorCollection<T> sub2 = new ManualSubscriberWithErrorCollection<T>(env);
+          env.subscribe(processor, sub2);
 
-      sub1.request(1);
-      expectRequest();
-      final T x = sendNextTFromUpstream();
-      expectNextElement(sub1, x);
-      sub1.request(1);
+          sub1.request(1);
+          expectRequest();
+          final T x = sendNextTFromUpstream();
+          expectNextElement(sub1, x);
+          sub1.request(1);
 
-      // sub1 has received one element, and has one demand pending
-      // sub2 has not yet requested anything
+          // sub1 has received one element, and has one demand pending
+          // sub2 has not yet requested anything
 
-      final Exception ex = new RuntimeException("Test exception");
-      sendError(ex);
-      sub1.expectError(ex);
-      sub2.expectError(ex);
+          final Exception ex = new RuntimeException("Test exception");
+          sendError(ex);
+          sub1.expectError(ex);
+          sub2.expectError(ex);
 
-      env.verifyNoAsyncErrors();
-    }};
+          env.verifyNoAsyncErrors();
+        }};
+
+      }
+    });
   }
 
   ////////////////////// SUBSCRIBER RULES VERIFICATION ///////////////////////////
@@ -580,49 +599,54 @@ public void spec317_mustSupportAPendingElementCountUpToLongMaxValue() throws Thr
 
   // A Processor
   //   must trigger `requestFromUpstream` for elements that have been requested 'long ago'
-  @Test
-  public void mustRequestFromUpstreamForElementsThatHaveBeenRequestedLongAgo() throws Exception {
-    new TestSetup(env, processorBufferSize) {{
-      ManualSubscriber<T> sub1 = newSubscriber();
-      sub1.request(20);
+  @Test @Subscribers(2)
+  public void mustRequestFromUpstreamForElementsThatHaveBeenRequestedLongAgo() throws Throwable {
+    optionalMultipleSubscribersTest(2, new Function<Long,TestSetup>() {
+      @Override
+      public TestSetup apply(Long subscribers) throws Throwable {
+        return new TestSetup(env, processorBufferSize) {{
+          ManualSubscriber<T> sub1 = newSubscriber();
+          sub1.request(20);
 
-      long totalRequests = expectRequest();
-      final T x = sendNextTFromUpstream();
-      expectNextElement(sub1, x);
+          long totalRequests = expectRequest();
+          final T x = sendNextTFromUpstream();
+          expectNextElement(sub1, x);
 
-      if (totalRequests == 1) {
-        totalRequests += expectRequest();
-      }
-      final T y = sendNextTFromUpstream();
-      expectNextElement(sub1, y);
+          if (totalRequests == 1) {
+            totalRequests += expectRequest();
+          }
+          final T y = sendNextTFromUpstream();
+          expectNextElement(sub1, y);
 
-      if (totalRequests == 2) {
-        totalRequests += expectRequest();
-      }
+          if (totalRequests == 2) {
+            totalRequests += expectRequest();
+          }
 
-      ManualSubscriber<T> sub2 = newSubscriber();
+          final ManualSubscriber<T> sub2 = newSubscriber();
 
-      // sub1 now has 18 pending
-      // sub2 has 0 pending
+          // sub1 now has 18 pending
+          // sub2 has 0 pending
 
-      final T z = sendNextTFromUpstream();
-      expectNextElement(sub1, z);
-      sub2.expectNone(); // since sub2 hasn't requested anything yet
+          final T z = sendNextTFromUpstream();
+          expectNextElement(sub1, z);
+          sub2.expectNone(); // since sub2 hasn't requested anything yet
 
-      sub2.request(1);
-      expectNextElement(sub2, z);
+          sub2.request(1);
+          expectNextElement(sub2, z);
 
-      if (totalRequests == 3) {
-        expectRequest();
-      }
+          if (totalRequests == 3) {
+            expectRequest();
+          }
 
-      // to avoid error messages during test harness shutdown
-      sendCompletion();
-      sub1.expectCompletion(env.defaultTimeoutMillis());
-      sub2.expectCompletion(env.defaultTimeoutMillis());
+          // to avoid error messages during test harness shutdown
+          sendCompletion();
+          sub1.expectCompletion(env.defaultTimeoutMillis());
+          sub2.expectCompletion(env.defaultTimeoutMillis());
 
-      env.verifyNoAsyncErrors();
-    }};
+          env.verifyNoAsyncErrors();
+        }};
+      }
+    });
   }
 
   /////////////////////// TEST INFRASTRUCTURE //////////////////////
@@ -635,6 +659,16 @@ public void notVerified(String message) {
     publisherVerification.notVerified(message);
   }
 
+  /**
+   * Test for feature that REQUIRES multiple subscribers to be supported by Publisher.
+   */
+  public void optionalMultipleSubscribersTest(long requiredSubscribersSupport, Function<Long, TestSetup> body) throws Throwable {
+    if (requiredSubscribersSupport > maxSupportedSubscribers())
+      notVerified("The Publisher under test only supports " +maxSupportedSubscribers()+ " subscribers, " +
+                      "while this test requires at least " + requiredSubscribersSupport + "to run.");
+    else body.apply(requiredSubscribersSupport);
+  }
+
   public abstract class TestSetup extends ManualPublisher<T> {
     final private ManualSubscriber<T> tees; // gives us access to an infinite stream of T values
     private Set<T> seenTees = new HashSet<T>();

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

@@ -4,6 +4,7 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.reactivestreams.tck.TestEnvironment.*;
+import org.reactivestreams.tck.support.Function;
 import org.reactivestreams.tck.support.Optional;
 import org.testng.SkipException;
 import org.testng.annotations.BeforeMethod;
@@ -808,8 +809,4 @@ public void notVerified(String message) {
     throw new SkipException(message);
   }
 
-  private interface Function<In, Out> {
-    public Out apply(In in) throws Throwable;
-  }
-
 }

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

@@ -667,7 +667,7 @@ public void expectCompletion(long timeoutMillis, String msg) throws InterruptedE
 
     @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
     public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, String requiredMessagePart) throws InterruptedException {
-      E err = expectError(expected);
+      final E err = expectError(expected);
       String message = err.getMessage();
       assertTrue(message.contains(requiredMessagePart),
         String.format("Got expected exception %s but missing message [%s], was: %s", err.getClass(), requiredMessagePart, expected));
@@ -680,7 +680,10 @@ public <E extends Throwable> E expectError(Class<E> expected) throws Interrupted
     @SuppressWarnings({"unchecked", "ThrowableResultOfMethodCallIgnored"})
     public <E extends Throwable> E expectError(Class<E> expected, long timeoutMillis) throws InterruptedException {
       error.expectCompletion(timeoutMillis, String.format("Subscriber %s did not call `registerOnError(%s)`", sub(), expected));
-      if (expected.isInstance(error.value())) {
+      if (error.value() == null) {
+        env.flop(String.format("Subscriber %s did not call `registerOnError(%s)`", sub(), expected));
+        return null; // make compiler happy
+      } else if (expected.isInstance(error.value())) {
         return (E) error.value();
       } else {
         env.flop(String.format("Subscriber %s called `registerOnError(%s)` rather than `registerOnError(%s)`", sub(), error.value(), expected));

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

@@ -0,0 +1,5 @@
+package org.reactivestreams.tck.support;
+
+public interface Function<In, Out> {
+  public Out apply(In in) throws Throwable;
+}