+tck fixes spec303 to not leak "still publishing" publisher

Since the Publisher could stay alive and keep publishing,
and keep getting demand in the recursion depth detecting test
it would exceed the buffer size by accumulating these onNext
signals, and then cause an error during *a different tests execution*
Now we do not accumulate these elements, and also set an explicit limit
on how long we try to blow up the recursive stack depth.

Stack depth is now also logged in as debug information which should
help implementers notice how their imple behave in this scenario.

The publisher is now always canceled at the end of this test,
in case it did not reach the max nr of elements it is allowed to
publish here, nor has it signalled completion.

Author: ktoso

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

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

tck/README.md

@@ -202,7 +202,7 @@ public class RangePublisherTest extends PublisherVerification<Integer> {
 }
 ```
 
-Note that hard-coded values *take precedence* over environment set values (!).
+Note that explicitly passed in values take precedence over values provided by the environment
 
 ## Subscriber Verification
 

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

@@ -124,28 +124,6 @@ public boolean skipStochasticTests() {
    */
   public abstract Processor<T, T> createIdentityProcessor(int bufferSize);
 
-  /**
-   * 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);
-
   /**
    * Return a Publisher that immediately signals {@code onError} to incoming subscriptions,
    * or {@code null} in order to skip them.

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

@@ -637,18 +637,33 @@ protected Long initialValue() {
           }
         };
 
+        final Latch runCompleted = new Latch(env);
+
         final ManualSubscriber<T> sub = new ManualSubscriberWithSubscriptionSupport<T>(env) {
+          // counts the number of signals received, used to break out from possibly infinite request/onNext loops
+          long signalsReceived = 0L;
+
           @Override
           public void onNext(T element) {
+            // NOT calling super.onNext as this test only cares about stack depths, not the actual values of elements
+            // which also simplifies this test as we do not have to drain the test buffer, which would otherwise be in danger of overflowing
+
+            signalsReceived += 1;
             stackDepthCounter.set(stackDepthCounter.get() + 1);
-            super.onNext(element);
+            env.debug(String.format("%s(recursion depth: %d)::onNext(%s)", this, stackDepthCounter.get(), element));
 
-            final Long callsUntilNow = stackDepthCounter.get();
+            final long callsUntilNow = stackDepthCounter.get();
             if (callsUntilNow > boundedDepthOfOnNextAndRequestRecursion()) {
               env.flop(String.format("Got %d onNext calls within thread: %s, yet expected recursive bound was %d",
                                      callsUntilNow, Thread.currentThread(), boundedDepthOfOnNextAndRequestRecursion()));
 
               // stop the recursive call chain
+              runCompleted.close();
+              return;
+            } else if (signalsReceived >= oneMoreThanBoundedLimit) {
+              // since max number of signals reached, and recursion depth not exceeded, we judge this as a success and
+              // stop the recursive call chain
+              runCompleted.close();
               return;
             }
 
@@ -657,14 +672,35 @@ public void onNext(T element) {
 
             stackDepthCounter.set(stackDepthCounter.get() - 1);
           }
+
+          @Override
+          public void onComplete() {
+            super.onComplete();
+            runCompleted.close();
+          }
+
+          @Override
+          public void onError(Throwable cause) {
+            super.onError(cause);
+            runCompleted.close();
+          }
         };
 
-        env.subscribe(pub, sub);
+        try {
+          env.subscribe(pub, sub);
 
-        sub.request(1); // kick-off the `request -> onNext -> request -> onNext -> ...`
+          sub.request(1); // kick-off the `request -> onNext -> request -> onNext -> ...`
 
-        sub.nextElementOrEndOfStream();
-        env.verifyNoAsyncErrors();
+          final String msg = String.format("Unable to validate call stack depth safety, " +
+                                               "awaited at-most %s signals (`maxOnNextSignalsInRecursionTest()`) or completion",
+                                           oneMoreThanBoundedLimit);
+          runCompleted.expectClose(env.defaultTimeoutMillis(), msg);
+          env.verifyNoAsyncErrors();
+        } finally {
+          // since the request/onNext recursive calls may keep the publisher running "forever",
+          // we MUST cancel it manually before exiting this test case
+          sub.cancel();
+        }
       }
     });
   }

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

@@ -38,9 +38,8 @@ public abstract class WithHelperPublisher<T> {
   /**
    * 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.
+   * By default an <b>asynchronously signalling Publisher</b> is provided, which will use {@link #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