+tck #181 provides default helperPublisher for subscriber tests

+ When subscriber verifications are extended using Integer
  we can use the existing publisher implementations from examples
  to drive the tests. This allows implementers who only care about
  implementing Subscribers to not care at all about implementing
  Publishers
+ In case the subscriber verification is extended using a custom signal
  type (defined as "not Integer") we do not magically generate a
  Publisher of this type, but instead fail the tests telling the
  implementer that he/she will need to implement a custom helperPublisher
  which is exactly what we have up until now always demanded from
  implementations.
+ We also provide tests to verify the test failures work as expected,
  and that the messages are indeed helpful.
+ Updated TCK README.md to clarify when to implement a custom publisher
+ fixed mistakenly added `<T>` on NumberPublisher - it is bound to Ints
+ TCK now depends on examples, as it uses the well documented async
  publisher to provide the default publisher for numbers. As well as
  in order to allow implementers to reuse those publishers when
  implementing custom Publishers to drive their Subscriber specs.
! Moved away from createHelper to createElement style of testing
  subscribers

Author: ktoso

.travis.yml

@@ -1,11 +1,11 @@
 language: java
 script:
-- ./gradlew check
+  - ./gradlew check
 cache:
   directories:
   - $HOME/.gradle
 jdk:
-- openjdk6
+  - openjdk6
 env:
   global:
   - TERM=dumb

examples/src/main/java/org/reactivestreams/example/unicast/AsyncSubscriber.java

@@ -1,6 +1,5 @@
 package org.reactivestreams.example.unicast;
 
-import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 

examples/src/main/java/org/reactivestreams/example/unicast/SyncSubscriber.java

@@ -1,6 +1,5 @@
 package org.reactivestreams.example.unicast;
 
-import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 

examples/src/test/java/org/reactivestreams/example/unicast/AsyncSubscriberTest.java

@@ -1,7 +1,5 @@
 package org.reactivestreams.example.unicast;
 
-import java.util.Collections;
-import java.util.Iterator;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.tck.SubscriberBlackboxVerification;
@@ -31,17 +29,11 @@ public AsyncSubscriberTest() {
 
   @Override public Subscriber<Integer> createSubscriber() {
     return new AsyncSubscriber<Integer>(e) {
-      private long acc;
       @Override protected boolean whenNext(final Integer element) {
         return true;
       }
     };
   }
-  @SuppressWarnings("unchecked")
-  @Override public Publisher<Integer> createHelperPublisher(long elements) {
-    if (elements > Integer.MAX_VALUE) return new InfiniteIncrementNumberPublisher(e);
-    else return new NumberIterablePublisher(0, (int)elements, e);
-  }
 
   @Test public void testAccumulation() throws InterruptedException {
 
@@ -64,4 +56,8 @@ public AsyncSubscriberTest() {
     latch.await(DefaultTimeoutMillis * 10, TimeUnit.MILLISECONDS);
     assertEquals(i.get(), 45);
   }
+
+  @Override public Integer createElement(int element) {
+    return element;
+  }
 }

examples/src/test/java/org/reactivestreams/example/unicast/SyncSubscriberTest.java

@@ -1,17 +1,14 @@
 package org.reactivestreams.example.unicast;
 
-import java.util.Collections;
-import java.util.Iterator;
-import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.tck.SubscriberBlackboxVerification;
 import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.AfterClass;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class SyncSubscriberTest extends SubscriberBlackboxVerification<Integer> {
@@ -39,9 +36,8 @@ public SyncSubscriberTest() {
       }
     };
   }
-  @SuppressWarnings("unchecked")
-  @Override public Publisher<Integer> createHelperPublisher(long elements) {
-    if (elements > Integer.MAX_VALUE) return new InfiniteIncrementNumberPublisher(e);
-    else return new NumberIterablePublisher(0, (int)elements, e);
+
+  @Override public Integer createElement(int element) {
+    return element;
   }
 }

tck/README.md

@@ -180,6 +180,36 @@ Subscriber rules Verification is split up into two files (styles) of tests.
 
 The Blackbox Verification tests do not require the implementation under test to be modified at all, yet they are *not* able to verify most rules. In Whitebox Verification, more control over `request()` calls etc. is required in order to validate rules more precisely.
 
+### Helper Publisher implementations
+Since testing a `Subscriber` is not possible without a corresponding `Publisher` the TCK Subscriber Verifications
+both provide a default "helper publisher" to drive its test and also alow to replace this Publisher with a custom implementation.
+
+For simple Subscribers which are able to consume elements of *any type*, it is **highly recommmended** to extend the
+SubscriberVerification (described below) classes providing the element type `java.lang.Integer`, like so: `... extends SubscriberBlackboxVerification<Integer>`.
+The reason for this is, that the TCK contains a default Publisher implementation which is able to signal `Integer` elements,
+thus alowing the implementer to strictly focus on only implementing a proper `Subscriber`, instead of having to implement
+an additional Publisher only in order to drive the Subscribers tests. This is especially important for library implementers
+which only want to implement a Subscriber – and do not want to spend time or thought on implementing a valid Publisher.
+
+If however any SubscriberVerification class is extended using a custom element type, e.g. like this `... extends SubscriberBlackboxVerification<Message>`,
+*the TCK will immediatly fail the entire subscriber test class* as it is unable to properly create signals of type `Message`
+(which can be some custom message type the `Subscriber` is able to consume). The exception thrown (`UnableToProvidePublisherException`)
+contains some information and directs the implementer towards implementing a custom helper publisher,
+which is done by overriding the `Publisher<T> createHelperPublisher(long elements)` method:
+
+```java
+@Override public Publisher<Message> createHelperPublisher(long elements) {
+  return new Publisher<Message>() { /* IMPL HERE */ };
+}
+```
+
+Summing up, we recommend implementing Subscribers which are able to consume any type of element, in this case the TCK
+should be driven using `Integer` elements as default publishers are already implemented for this type. If the
+`Subscriber` is unable to consume `Integer` elements, the implementer MUST implement a custom `Publisher<T>` that will
+be able to signal the required element types. It is of course both possible and recommended to re-use existing
+implemenations (which can be seen in the examples sub-project) to create these custom Publishers – an example of
+such re-use can be found in [ProvidedHelperPublisherForSubscriberVerificationTest#createStringPublisher](https://github.com/reactive-streams/reactive-streams/blob/master/tck/src/test/java/org/reactivestreams/tck/ProvidedHelperPublisherForSubscriberVerificationTest.java#L215)
+
 ### Subscriber Blackbox Verification
 
 Blackbox Verification does not require any additional work except from providing a `Subscriber` and `Publisher` instances to the TCK:
@@ -205,11 +235,6 @@ public class MySubscriberBlackboxVerificationTest extends SubscriberBlackboxVeri
   public Subscriber<Integer> createSubscriber() {
     return new MySubscriber<Integer>();
   }
-
-  @Override
-  public Publisher<Integer> createHelperPublisher(long elements) {
-    return new MyRangePublisher<Integer>(1, elements);
-  }
 }
 ```
 
@@ -285,11 +310,6 @@ public class MySubscriberWhiteboxVerificationTest extends SubscriberWhiteboxVeri
     };
   }
 
-  @Override
-  public Publisher<Integer> createHelperPublisher(long elements) {
-    return new MyRangePublisher<Integer>(1, elements);
-  }
-
 }
 ```
 

tck/build.gradle

@@ -2,5 +2,6 @@ description = 'reactive-streams-tck'
 dependencies {
     compile group: 'org.testng', name: 'testng', version:'5.14.10'
     compile project(':reactive-streams')
+    compile project(':reactive-streams-examples')
 }
 test.useTestNG()

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

@@ -17,7 +17,7 @@
 import java.util.HashSet;
 import java.util.Set;
 
-public abstract class IdentityProcessorVerification<T> implements SubscriberWhiteboxVerificationRules, PublisherVerificationRules {
+public abstract class IdentityProcessorVerification<T> extends WithHelperPublisher<T> implements SubscriberWhiteboxVerificationRules, PublisherVerificationRules {
 
   private final TestEnvironment env;
 
@@ -65,6 +65,10 @@ public Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe) {
         return IdentityProcessorVerification.this.createSubscriber(probe);
       }
 
+      @Override public T createElement(int element) {
+        return IdentityProcessorVerification.this.createElement(element);
+      }
+
       @Override
       public Publisher<T> createHelperPublisher(long elements) {
         return IdentityProcessorVerification.this.createHelperPublisher(elements);
@@ -109,15 +113,24 @@ public boolean skipStochasticTests() {
   public abstract Processor<T, T> createIdentityProcessor(int bufferSize);
 
   /**
-   * Helper method required for running the Publisher rules against a Publisher.
-   * It must create a Publisher for a stream with exactly the given number of elements.
-   * If {@code elements} is {@code Long.MAX_VALUE} the produced stream must be infinite.
-   *
-   * The stream must not produce the same element twice (in case of an infinite stream this requirement
-   * is relaxed to only apply to the elements that are actually requested during all tests).
-   *
-   * @param elements exact number of elements this publisher should emit,
-   *                 unless equal to {@code Long.MAX_VALUE} in which case the stream should be effectively infinite
+   * Helper method required for creating the Publisher to which the tested Subscriber will be subscribed and tested against.
+   * <p>
+   * By default an <b>asynchronously signalling Publisher</b> is provided, which will use
+   * {@link org.reactivestreams.tck.SubscriberBlackboxVerification#createElement(int)} to generate elements type
+   * your Subscriber is able to consume.
+   * <p>
+   * Sometimes you may want to implement your own custom custom helper Publisher - to validate behaviour of a Subscriber
+   * when facing a synchronous Publisher for example. If you do, it MUST emit the exact number of elements asked for
+   * (via the {@code elements} parameter) and MUST also must treat the following numbers of elements in these specific ways:
+   * <ul>
+   *   <li>
+   *     If {@code elements} is {@code Long.MAX_VALUE} the produced stream must be infinite.
+   *   </li>
+   *   <li>
+   *     If {@code elements} is {@code 0} the {@code Publisher} should signal {@code onComplete} immediatly.
+   *     In other words, it should represent a "completed stream".
+   *   </li>
+   * </ul>
    */
   public abstract Publisher<T> createHelperPublisher(long elements);
 

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

@@ -9,9 +9,13 @@
 import org.reactivestreams.tck.support.TestException;
 import org.reactivestreams.tck.support.SubscriberBlackboxVerificationRules;
 import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import static org.reactivestreams.tck.SubscriberWhiteboxVerification.BlackboxSubscriberProxy;
 
 /**
@@ -25,36 +29,32 @@
  * @see org.reactivestreams.Subscriber
  * @see org.reactivestreams.Subscription
  */
-public abstract class SubscriberBlackboxVerification<T> implements SubscriberBlackboxVerificationRules {
+public abstract class SubscriberBlackboxVerification<T> extends WithHelperPublisher<T> implements SubscriberBlackboxVerificationRules {
 
   private final TestEnvironment env;
 
   protected SubscriberBlackboxVerification(TestEnvironment env) {
     this.env = env;
   }
 
+  // USER API
+
   /**
    * This is the main method you must implement in your test incarnation.
    * It must create a new {@link org.reactivestreams.Subscriber} instance to be subjected to the testing logic.
    */
   public abstract Subscriber<T> createSubscriber();
 
+  // ENV SETUP
+
   /**
-   * Helper method required for generating test elements.
-   * It must create a {@link org.reactivestreams.Publisher} for a stream with exactly the given number of elements.
-   * <p>
-   * It also must treat the following numbers of elements in these specific ways:
-   * <ul>
-   *   <li>
-   *    If {@code elements} is {@code Long.MAX_VALUE} the produced stream must be infinite.
-   *   </li>
-   *   <li>
-   *    If {@code elements} is {@code 0} the {@code Publisher} should signal {@code onComplete} immediatly.
-   *    In other words, it should represent a "completed stream".
-   *   </li>
-   * </ul>
+   * Executor service used by the default provided asynchronous Publisher.
+   * @see #createHelperPublisher(long)
    */
-  public abstract Publisher<T> createHelperPublisher(long elements);
+  private ExecutorService publisherExecutor;
+  @BeforeClass public void startPublisherExecutorService() { publisherExecutor = Executors.newFixedThreadPool(4); }
+  @AfterClass public void shutdownPublisherExecutorService() { if (publisherExecutor != null) publisherExecutor.shutdown(); }
+  @Override public ExecutorService publisherExecutorService() { return publisherExecutor; }
 
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
 

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

@@ -4,14 +4,18 @@
 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.reactivestreams.tck.support.TestException;
 import org.reactivestreams.tck.support.SubscriberWhiteboxVerificationRules;
 import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
@@ -21,14 +25,16 @@
  * @see org.reactivestreams.Subscriber
  * @see org.reactivestreams.Subscription
  */
-public abstract class SubscriberWhiteboxVerification<T> implements SubscriberWhiteboxVerificationRules {
+public abstract class SubscriberWhiteboxVerification<T> extends WithHelperPublisher<T> implements SubscriberWhiteboxVerificationRules {
 
   private final TestEnvironment env;
 
   protected SubscriberWhiteboxVerification(TestEnvironment env) {
     this.env = env;
   }
 
+  // USER API
+
   /**
    * This is the main method you must implement in your test incarnation.
    * It must create a new {@link org.reactivestreams.Subscriber} instance to be subjected to the testing logic.
@@ -38,33 +44,16 @@ protected SubscriberWhiteboxVerification(TestEnvironment env) {
    */
   public abstract Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe);
 
-  /**
-   * Helper method required for generating test elements.
-   * It must create a {@link org.reactivestreams.Publisher} for a stream with exactly the given number of elements.
-   * <p>
-   * It also must treat the following numbers of elements in these specific ways:
-   * <ul>
-   *   <li>
-   *    If {@code elements} is {@code Long.MAX_VALUE} the produced stream must be infinite.
-   *   </li>
-   *   <li>
-   *    If {@code elements} is {@code 0} the {@code Publisher} should signal {@code onComplete} immediatly.
-   *    In other words, it should represent a "completed stream".
-   *   </li>
-   * </ul>
-   */
-  public abstract Publisher<T> createHelperPublisher(long elements);
+  // ENV SETUP
 
   /**
-   * Used to break possibly infinite wait-loops.
-   * Some Rules use the "eventually stop signalling" wording, which requires the test to spin accepting {@code onNext}
-   * signals until no more are signalled. In these tests, this value will be used as upper bound on the number of spin iterations.
-   *
-   * Override this method in case your implementation synchronously signals very large batches before reacting to cancellation (for example).
+   * Executor service used by the default provided asynchronous Publisher.
+   * @see #createHelperPublisher(long)
    */
-  public long maxOnNextSignalsInTest() {
-    return 100;
-  }
+  private ExecutorService publisherExecutor;
+  @BeforeClass public void startPublisherExecutorService() { publisherExecutor = Executors.newFixedThreadPool(4); }
+  @AfterClass public void shutdownPublisherExecutorService() { if (publisherExecutor != null) publisherExecutor.shutdown(); }
+  @Override public ExecutorService publisherExecutorService() { return publisherExecutor; }
 
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////