+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

@@ -60,4 +60,8 @@ public AsyncSubscriberTest() {
   @Override public Integer createElement(int element) {
     return element;
   }
+
+  @Override public Publisher<Integer> createHelperPublisher(long elements) {
+    return super.createHelperPublisher(elements);
+  }
 }

tck/README.md

@@ -180,34 +180,28 @@ The Blackbox Verification tests do not require the implementation under test to
 
 ### 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:
+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.
+
+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. The simplest valid implemenation is to return the incoming
+`id` *as the element* in a verification using `Integer`s as element types. 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:

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

@@ -15,6 +15,7 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.reactivestreams.tck.Annotations.NotVerified;
 import static org.reactivestreams.tck.Annotations.Required;
@@ -224,7 +225,24 @@ public void spec209_blackbox_mustBePreparedToReceiveAnOnCompleteSignalWithPreced
     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/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 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;
           }