+tck 2.9 blackbox verification should fully drive publisher

It should not use the helper publisher as it may signal asynchronously
as well as onComplete in undefined ways - we need to control the
publisher in this case because of the very specific test scenario

Author: ktoso

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

@@ -17,14 +17,14 @@
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class AsyncSubscriberTest extends SubscriberBlackboxVerification<Integer> {
-  final static long DefaultTimeoutMillis = 100;
+  final static long DEFAULT_TIMEOUT_MILLIS = 100;
 
   private ExecutorService e;
   @BeforeClass void before() { e = Executors.newFixedThreadPool(4); }
   @AfterClass void after() { if (e != null) e.shutdown(); }
 
   public AsyncSubscriberTest() {
-    super(new TestEnvironment(DefaultTimeoutMillis));
+    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS));
   }
 
   @Override public Subscriber<Integer> createSubscriber() {
@@ -53,11 +53,15 @@ public AsyncSubscriberTest() {
     };
 
     new NumberIterablePublisher(0, 10, e).subscribe(sub);
-    latch.await(DefaultTimeoutMillis * 10, TimeUnit.MILLISECONDS);
+    latch.await(DEFAULT_TIMEOUT_MILLIS * 10, TimeUnit.MILLISECONDS);
     assertEquals(i.get(), 45);
   }
 
   @Override public Integer createElement(int element) {
     return element;
   }
+
+  @Override public Publisher<Integer> createHelperPublisher(long elements) {
+    return super.createHelperPublisher(elements);
+  }
 }

tck/README.md

@@ -51,7 +51,7 @@ Specification rule abides the following naming convention: `spec###_DESC` where:
 ```
 
 The prefixes of the names of the test methods are used in order to signify the character of the test. For example, these are the kinds of prefixes you may find:
-"required_", "optional_", "stochastic_", "untested_".
+"`required_`", "`optional_`", "`stochastic_`", "`untested_`".
 
 Explanations:
 
@@ -180,36 +180,44 @@ 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
+### createElement and 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.
+both provide a default "*helper publisher*" to drive its test and also alow to replace this Publisher with a custom implementation.
+The helper publisher is an asynchronous publisher by default - meaning that your subscriber can not blindly assume single threaded execution.
 
-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.
+While the `Publisher` implementation is provided, creating the signal elements is not – this is because a given Subscriber
+may for example only work with `HashedMessage` or some other specific kind of signal. The TCK is unable to generate such
+special messages automatically, so we provide the `T createElement(Integer id)` method to be implemented as part of
+Subscriber Verifications which should take the given ID and return an element of type `T` (where `T` is the type of
+elements flowing into the `Subscriber<T>`, as known thanks to `... extends WhiteboxSubscriberVerification<T>`) representing
+an element of the stream that will be passed on to the Subscriber.
 
-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:
+The simplest valid implemenation is to return the incoming `id` *as the element* in a verification using `Integer`s as element types:
+
+```java
+public class MySubscriberTest extends SubscriberBlackboxVerification<Integer> {
+
+  // ...
+
+  @Override
+  public Integer createElement(int element) { return element; }
+}
+```
+
+
+The `createElement` method MAY be called from multiple
+threads, so in case of more complicated implementations, please be aware of this fact.
+
+**Very Advanced**: While we do not expect many implementations having to do so, it is possible to take full control of the `Publisher`
+which will be driving the TCKs test. You can do this by implementing the `createHelperPublisher` method in which you can implement your
+own Publisher which will then be used by the TCK to drive your Subscriber tests:
 
 ```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:
@@ -235,6 +243,11 @@ public class MySubscriberBlackboxVerificationTest extends SubscriberBlackboxVeri
   public Subscriber<Integer> createSubscriber() {
     return new MySubscriber<Integer>();
   }
+
+  @Override
+  public Integer createElement(int element) {
+    return element;
+  }
 }
 ```
 
@@ -310,6 +323,11 @@ public class MySubscriberWhiteboxVerificationTest extends SubscriberWhiteboxVeri
     };
   }
 
+  @Override
+  public Integer createElement(int element) {
+    return element;
+  }
+
 }
 ```
 

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

@@ -4,7 +4,6 @@
 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;

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

@@ -16,6 +16,7 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.reactivestreams.tck.SubscriberWhiteboxVerification.BlackboxSubscriberProxy;
 
@@ -224,7 +225,24 @@ public void required_spec209_blackbox_mustBePreparedToReceiveAnOnCompleteSignalW
     blackboxSubscriberWithoutSetupTest(new BlackboxTestStageTestRun() {
       @Override
       public void run(BlackboxTestStage stage) throws Throwable {
-        final Publisher<T> pub = createHelperPublisher(0);
+        final Publisher<T> pub = new Publisher<T>() {
+          @Override public void subscribe(final Subscriber<? super T> s) {
+            s.onSubscribe(new Subscription() {
+              private boolean completed = false;
+
+              @Override public void request(long n) {
+                if (!completed) {
+                  completed = true;
+                  s.onComplete(); // Publisher now realises that it is in fact already completed
+                }
+              }
+
+              @Override public void cancel() {
+                // noop, ignore
+              }
+            });
+          }
+        };
 
         final Subscriber<T> sub = createSubscriber();
         final BlackboxSubscriberProxy<T> probe = stage.createBlackboxSubscriberProxy(env, sub);

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

@@ -27,6 +27,9 @@ public abstract class WithHelperPublisher<T> {
    * Sometimes the Subscriber may be limited in what type of element it is able to consume, this you may have to implement
    * this method such that the emitted element matches the Subscribers requirements. Simplest implementations would be
    * to simply pass in the {@code element} as payload of your custom element, such as appending it to a String or other identifier.
+   * <p/>
+   * <b>Warning:</b> This method may be called concurrently by the helper publisher, thus it should be implemented in a
+   * thread-safe manner.
    *
    * @return element of the matching type {@code T} that will be delivered to the tested Subscriber
    */

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

@@ -22,8 +22,7 @@ public HelperPublisher(final int from, final int to, final Function<Integer, T>
                 else try {
                   return create.apply(at++);
                 } catch (Throwable t) {
-                  throw new HelperPublisherException(
-                    String.format("Failed to create element in %s for id %s!", getClass().getSimpleName(), at - 1), t);
+                  throw new IllegalStateException(String.format("Failed to create element for id %d!", at - 1), t);
                 }
               }
               @Override public void remove() { throw new UnsupportedOperationException(); }

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

@@ -18,7 +18,7 @@ public InfiniteHelperPublisher(final Function<Integer, T> create, final Executor
                 try {
                   return create.apply(at++); // Wraps around on overflow
                 } catch (Throwable t) {
-                  throw new HelperPublisherException(
+                  throw new IllegalStateException(
                     String.format("Failed to create element in %s for id %s!", getClass().getSimpleName(), at - 1), t);
                 }
               }

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

@@ -28,22 +28,20 @@ public class IdentityProcessorVerificationTest extends TCKVerificationSupport {
   public void required_spec104_mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError_shouldBeIgnored() throws Throwable {
     requireTestSkip(new ThrowingRunnable() {
       @Override public void run() throws Throwable {
-        new IdentityProcessorVerification(newTestEnvironment(), DEFAULT_TIMEOUT_MILLIS){
-          @Override public Processor createIdentityProcessor(int bufferSize) {
+        new IdentityProcessorVerification<Integer>(newTestEnvironment(), DEFAULT_TIMEOUT_MILLIS){
+          @Override public Processor<Integer, Integer> createIdentityProcessor(int bufferSize) {
             return new NoopProcessor();
           }
 
           @Override public ExecutorService publisherExecutorService() { return ex; }
 
-          @Override public Object createElement(int element) {
-            return null;
-          }
+          @Override public Integer createElement(int element) { return element; }
 
-          @Override public Publisher createHelperPublisher(long elements) {
+          @Override public Publisher<Integer> createHelperPublisher(long elements) {
             return SKIP;
           }
 
-          @Override public Publisher createErrorStatePublisher() {
+          @Override public Publisher<Integer> createErrorStatePublisher() {
             return SKIP;
           }