+tck #181 provides default helperPublisher for subscriber tests

ktoso · ktoso · commit 9efb1413058c · 2014-12-27T18:09:45.000+01:00
+ 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 `&lt;T&gt;` 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.
diff --git a/.travis.yml b/.travis.yml
@@ -1,11 +1,11 @@
 language: java
 script:
-- ./gradlew check
+  - ./gradlew check
 cache:
   directories:
   - $HOME/.gradle
 jdk:
-- openjdk6
+  - openjdk6
 env:
   global:
   - TERM=dumb
diff --git a/examples/src/main/java/org/reactivestreams/example/unicast/AsyncSubscriber.java b/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;
 
diff --git a/examples/src/main/java/org/reactivestreams/example/unicast/SyncSubscriber.java b/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;
 
diff --git a/examples/src/test/java/org/reactivestreams/example/unicast/AsyncSubscriberTest.java b/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 {
 
@@ -60,7 +52,7 @@ public AsyncSubscriberTest() {
       }
     };
 
-    new NumberIterablePublisher<Integer>(0, 10, e).subscribe(sub);
+    new NumberIterablePublisher(0, 10, e).subscribe(sub);
     latch.await(DefaultTimeoutMillis * 10, TimeUnit.MILLISECONDS);
     assertEquals(i.get(), 45);
   }
diff --git a/examples/src/test/java/org/reactivestreams/example/unicast/SyncSubscriberTest.java b/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,5 @@ 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);
-  }
+
 }
diff --git a/tck/README.md b/tck/README.md
@@ -178,6 +178,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:
@@ -203,11 +233,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);
-  }
 }
 ```
 
@@ -283,11 +308,6 @@ public class MySubscriberWhiteboxVerificationTest extends SubscriberWhiteboxVeri
     };
   }
 
-  @Override
-  public Publisher<Integer> createHelperPublisher(long elements) {
-    return new MyRangePublisher<Integer>(1, elements);
-  }
-
 }
 ```
 
diff --git a/tck/build.gradle b/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()
diff --git a/tck/src/main/java/org/reactivestreams/tck/SubscriberBlackboxVerification.java b/tck/src/main/java/org/reactivestreams/tck/SubscriberBlackboxVerification.java
@@ -3,14 +3,24 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.reactivestreams.example.unicast.InfiniteIncrementNumberPublisher;
+import org.reactivestreams.example.unicast.NumberIterablePublisher;
 import org.reactivestreams.tck.TestEnvironment.ManualPublisher;
 import org.reactivestreams.tck.TestEnvironment.ManualSubscriber;
 import org.reactivestreams.tck.support.Optional;
 import org.reactivestreams.tck.support.TestException;
+import org.reactivestreams.tck.support.UnableToProvidePublisherException;
 import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.BeforeTest;
 import org.testng.annotations.Test;
 
+import java.lang.reflect.ParameterizedType;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import static org.reactivestreams.tck.Annotations.NotVerified;
 import static org.reactivestreams.tck.Annotations.Required;
 import static org.reactivestreams.tck.SubscriberWhiteboxVerification.BlackboxSubscriberProxy;
@@ -34,28 +44,80 @@ 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
+
+  /**
+   * Executor service used by the default provided asynchronous Publisher.
+   * @see #createHelperPublisher(long)
+   */
+  private ExecutorService publisherExecutor;
+  @BeforeClass public void startPublisherExecutorService() { publisherExecutor = Executors.newFixedThreadPool(4); }
+  @AfterClass public void shutdownPublisherExecutorService() { if (publisherExecutor != null) publisherExecutor.shutdown(); }
+  public ExecutorService publisherExecutorService() { return publisherExecutor; }
+
+  /**
+   * Returns the Class of the {@code <T>} type parameter with which this SubscriberVerification was instanciated.
+   * This is used to validate if the default provided helper {@link Publisher} can drive the tests for the tested
+   * {@link Subscriber} or not.
+   *
+   * @see #createHelperPublisher(long)
+   */
+  @SuppressWarnings("unchecked")
+  public Class<T> elementClass() {
+    return (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
+  }
+
   /**
-   * 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.
+   * Validates if the required element type is able to be signalled by the default Publisher.
    * <p>
-   * It also must treat the following numbers of elements in these specific ways:
+   * If it is able to create such Publisher returns {@code true},
+   * otherwise throws an {@link org.reactivestreams.tck.support.UnableToProvidePublisherException}.
+   */
+  @BeforeTest public boolean validateAbilityToProvidePublisher() {
+    if (!elementClass().isAssignableFrom(Integer.class)) {
+      throw new UnableToProvidePublisherException(String.format("Unable to provide Publisher of `%s` elements (can only emit `java.lang.Integer`)!", elementClass().getCanonicalName()));
+    } else return true;
+  }
+
+  /**
+   * Helper method required for creating the Publisher to which the tested Subscriber will be subscribed and tested against.
+   * <p>
+   * By default an <b>Asynchronous Integer signalling Publisher</b> is provided, however if this SubscriberVerification
+   * was created using a different type (and your Subscriber is only able to consume those signals (e.g. "MyMessage"))
+   * the default provided Publisher will NOT be able to generate those messages and will fail the entire test class and
+   * asking you to provide a custom {@link Publisher} which is able to generate elements of type {@code T}.
+   * <p>
+   * When implementing a custom helper Publisher 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.
+   *     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".
+   *     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);
+  @SuppressWarnings("unchecked")
+  public Publisher<T> createHelperPublisher(long elements) {
+    if (validateAbilityToProvidePublisher()) {
+      // we can safely provide the Integer based Publishers
+      if (elements > Integer.MAX_VALUE) {
+        return (Publisher<T>) new InfiniteIncrementNumberPublisher(publisherExecutorService());
+      } else {
+        return (Publisher<T>) new NumberIterablePublisher(0, (int)elements, publisherExecutorService());
+      }
+    } else throw new RuntimeException("Unreachable. `validateAbilityToProvidePublisher` should have thrown!");
+  }
 
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
 
diff --git a/tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java b/tck/src/main/java/org/reactivestreams/tck/SubscriberWhiteboxVerification.java
@@ -3,15 +3,25 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.reactivestreams.example.unicast.InfiniteIncrementNumberPublisher;
+import org.reactivestreams.example.unicast.NumberIterablePublisher;
 import org.reactivestreams.tck.Annotations.Additional;
 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.UnableToProvidePublisherException;
 import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.BeforeTest;
 import org.testng.annotations.Test;
 
+import java.lang.reflect.ParameterizedType;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import static org.reactivestreams.tck.Annotations.NotVerified;
 import static org.reactivestreams.tck.Annotations.Required;
 import static org.testng.Assert.assertEquals;
@@ -31,6 +41,8 @@ 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.
@@ -40,32 +52,71 @@ protected SubscriberWhiteboxVerification(TestEnvironment env) {
    */
   public abstract Subscriber<T> createSubscriber(WhiteboxSubscriberProbe<T> probe);
 
+  // ENV SETUP
+
+  /**
+   * Executor service used by the default provided asynchronous Publisher.
+   * @see #createHelperPublisher(long)
+   */
+  private ExecutorService publisherExecutor;
+  @BeforeClass public void startPublisherExecutorService() { publisherExecutor = Executors.newFixedThreadPool(4); }
+  @AfterClass public void shutdownPublisherExecutorService() { if (publisherExecutor != null) publisherExecutor.shutdown(); }
+  public ExecutorService publisherExecutorService() { return publisherExecutor; }
+
   /**
-   * 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.
+   * Returns the Class of the {@code <T>} type parameter with which this SubscriberVerification was instanciated.
+   * This is used to validate if the default provided helper {@link Publisher} can drive the tests for the tested
+   * {@link Subscriber} or not.
+   *
+   * @see #createHelperPublisher(long)
+   */
+  @SuppressWarnings("unchecked")
+  public Class<T> elementClass() {
+    return (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
+  }
+
+  /**
+   * Validates if the required element type is able to be signalled by the default Publisher.
    * <p>
-   * It also must treat the following numbers of elements in these specific ways:
+   * If it is able to create such Publisher returns {@code true},
+   * otherwise throws an {@link org.reactivestreams.tck.support.UnableToProvidePublisherException}.
+   */
+  @BeforeTest public boolean validateAbilityToProvidePublisher() {
+    if (!elementClass().isAssignableFrom(Integer.class)) {
+      throw new UnableToProvidePublisherException(String.format("Unable to provide Publisher of `%s` elements (can only emit `java.lang.Integer`)!", elementClass().getCanonicalName()));
+    } else return true;
+  }
+
+  /**
+   * Helper method required for creating the Publisher to which the tested Subscriber will be subscribed and tested against.
+   * <p>
+   * By default an <b>Asynchronous Integer signalling Publisher</b> is provided, however if this SubscriberVerification
+   * was created using a different type (and your Subscriber is only able to consume those signals (e.g. "MyMessage"))
+   * the default provided Publisher will NOT be able to generate those messages and will fail the entire test class and
+   * asking you to provide a custom {@link Publisher} which is able to generate elements of type {@code T}.
+   * <p>
+   * When implementing a custom helper Publisher 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.
+   *     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".
+   *     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);
-
-  /**
-   * 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).
-   */
-  public long maxOnNextSignalsInTest() {
-    return 100;
+  @SuppressWarnings("unchecked")
+  public Publisher<T> createHelperPublisher(long elements) {
+    if (validateAbilityToProvidePublisher()) {
+      // we can safely provide the Integer based Publishers
+      if (elements > Integer.MAX_VALUE) {
+        return (Publisher<T>) new InfiniteIncrementNumberPublisher(publisherExecutorService());
+      } else {
+        return (Publisher<T>) new NumberIterablePublisher(0, (int)elements, publisherExecutorService());
+      }
+    } else throw new RuntimeException("Unreachable. `validateAbilityToProvidePublisher` should have thrown!");
   }
 
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
diff --git a/tck/src/main/java/org/reactivestreams/tck/support/UnableToProvidePublisherException.java b/tck/src/main/java/org/reactivestreams/tck/support/UnableToProvidePublisherException.java
diff --git a/tck/src/test/java/org/reactivestreams/tck/ProvidedHelperPublisherForSubscriberVerificationTest.java b/tck/src/test/java/org/reactivestreams/tck/ProvidedHelperPublisherForSubscriberVerificationTest.java

Original file line number	Diff line number	Diff line change
`@@ -2,5 +2,6 @@ description = 'reactive-streams-tck'`
`2`	`2`	`dependencies {`
`3`	`3`	`compile group: 'org.testng', name: 'testng', version:'5.14.10'`
`4`	`4`	`compile project(':reactive-streams')`
	`5`	`+ compile project(':reactive-streams-examples')`
`5`	`6`	`}`
`6`	`7`	`test.useTestNG()`