+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

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

@@ -124,6 +124,15 @@ public long boundedDepthOfOnNextAndRequestRecursion() {
     return 1;
   }
 
+  /**
+   * Used to break possibly infinite request/onNext loops.
+   *
+   * Override this method in case your implementation synchronously signals very large batches before reacting to cancellation.
+   */
+  public long maxOnNextSignalsInReccursionTest() {
+    return 1000;
+  }
+
   /**
    * The amount of time after which a cancelled Subscriber reference should be dropped.
    * See Rule 3.13 for details.
@@ -637,18 +646,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 > maxOnNextSignalsInReccursionTest()) {
+              // 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 +681,33 @@ 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);
 
         sub.request(1); // kick-off the `request -> onNext -> request -> onNext -> ...`
 
-        sub.nextElementOrEndOfStream();
+        final String msg = String.format("Unable to validate call stack depth safety, " +
+                                             "awaited at-most %s signals (`maxOnNextSignalsInReccursionTest()`) or completion",
+                                         maxOnNextSignalsInReccursionTest());
+        runCompleted.expectClose(env.defaultTimeoutMillis(), msg);
         env.verifyNoAsyncErrors();
+
+        // 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/SubscriberWhiteboxVerification.java

@@ -4,10 +4,9 @@
 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.reactivestreams.tck.support.TestException;
 import org.testng.SkipException;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;

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

@@ -538,7 +538,7 @@ public BlackholeSubscriberWithSubscriptionSupport(TestEnvironment env) {
 
     @Override
     public void onNext(T element) {
-      env.debug(String.format("%s::onNext(%s)", this, element));
+      env.debug(String.format("%s(blackhole)::onNext(%s)", this, element));
       if (!subscription.isCompleted()) {
         env.flop(String.format("Subscriber::onNext(%s) called before Subscriber::onSubscribe", element));
       }