+tck reactive-streams#308 allow configuring "no events during N time"…

… separately

.travis.yml

@@ -11,6 +11,7 @@ env:
   global:
   - TERM=dumb
   - DEFAULT_TIMEOUT_MILLIS=300
+  - DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS=200
   - PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=300
 addons:
   # Fix OpenJDK build. Issue: https://github.com/travis-ci/travis-ci/issues/5227 & https://docs.travis-ci.com/user/hostname

tck/README.md

@@ -209,25 +209,32 @@ within the TCK which await for something to happen. The other timeout is `publis
 [Rule 3.13](https://github.com/reactive-streams/reactive-streams-jvm#3.13) which defines that `Subscriber` references MUST be dropped
 by the Publisher.
 
+Note that the TCK differenciates between timeouts for "waiting for a signal" (``defaultTimeoutMillis``),
+and "asserting no signals happen during a given amount of time" (``envDefaultNoSignalsTimeoutMillis``).
+While the latter defaults to the prior, it may be useful to tweak them independently when running on continious 
+integration servers (for example, keeping the no-signals timeout significantly lower).
+
 In order to configure these timeouts (for example when running on a slow continious integtation machine), you can either:
 
 **Use env variables** to set these timeouts, in which case the you can do:
 
 ```bash
-export DEFAULT_TIMEOUT_MILLIS=300
-export PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=500
+export DEFAULT_TIMEOUT_MILLIS=100
+export DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS=100
+export PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=300
 ```
 
 Or **define the timeouts explicitly in code**:
 
 ```java
 public class RangePublisherTest extends PublisherVerification<Integer> {
 
-  public static final long DEFAULT_TIMEOUT_MILLIS = 300L;
+  public static final long DEFAULT_TIMEOUT_MILLIS = 100L;
+  public static final long DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS = DEFAULT_TIMEOUT_MILLIS;
   public static final long PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 500L;
 
   public RangePublisherTest() {
-    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS);
+    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS, DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS);
   }
 
   // ...

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

@@ -79,7 +79,7 @@ public static long envPublisherReferenceGCTimeoutMillis() {
     if (envMillis == null) return DEFAULT_PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS;
     else try {
       return Long.parseLong(envMillis);
-    } catch(NumberFormatException ex) {
+    } catch (NumberFormatException ex) {
       throw new IllegalArgumentException(String.format("Unable to parse %s env value [%s] as long!", PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS_ENV, envMillis), ex);
     }
   }
@@ -368,7 +368,7 @@ public void onError(Throwable cause) {
           onSubscribeLatch.expectClose("Should have received onSubscribe");
           onErrorlatch.expectClose(String.format("Error-state Publisher %s did not call `onError` on new Subscriber", pub));
 
-          env.verifyNoAsyncErrors(env.defaultTimeoutMillis());
+          env.verifyNoAsyncErrors();
           }
       });
     } catch (SkipException se) {
@@ -699,8 +699,7 @@ public void onNext(T element) {
 
         env.subscribe(pub, sub);
 
-        long delay = env.defaultTimeoutMillis();
-        env.verifyNoAsyncErrors(delay);
+        env.verifyNoAsyncErrors();
       }
     });
   }
@@ -1060,7 +1059,7 @@ public void onNext(T element) {
 
         // no onError should be signalled
         try {
-          env.verifyNoAsyncErrors(env.defaultTimeoutMillis());
+          env.verifyNoAsyncErrors();
         } finally {
           sub.cancel();
         }

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

@@ -711,7 +711,7 @@ public void expectError(Throwable expected, long timeoutMillis) throws Interrupt
     }
 
     public void expectNone() throws InterruptedException {
-      expectNone(env.defaultTimeoutMillis());
+      expectNone(env.defaultNoSignalsTimeoutMillis());
     }
 
     public void expectNone(long withinMillis) throws InterruptedException {

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

@@ -22,7 +22,10 @@ public class TestEnvironment {
   private static final String DEFAULT_TIMEOUT_MILLIS_ENV = "DEFAULT_TIMEOUT_MILLIS";
   private static final long DEFAULT_TIMEOUT_MILLIS = 100;
 
+  private static final String DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS_ENV = "DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS";
+
   private final long defaultTimeoutMillis;
+  private final long defaultNoSignalsTimeoutMillis;
   private final boolean printlnDebug;
 
   private CopyOnWriteArrayList<Throwable> asyncErrors = new CopyOnWriteArrayList<Throwable>();
@@ -32,16 +35,29 @@ public class TestEnvironment {
    * interactions. Longer timeout does not invalidate the correctness of
    * the implementation, but can in some cases result in longer time to
    * run the tests.
-   *
    * @param defaultTimeoutMillis default timeout to be used in all expect* methods
+   * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore
    * @param printlnDebug         if true, signals such as OnNext / Request / OnComplete etc will be printed to standard output,
-   *                             often helpful to pinpoint simple race conditions etc.
    */
-  public TestEnvironment(long defaultTimeoutMillis, boolean printlnDebug) {
+  public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, boolean printlnDebug) {
     this.defaultTimeoutMillis = defaultTimeoutMillis;
+    this.defaultNoSignalsTimeoutMillis = defaultNoSignalsTimeoutMillis;
     this.printlnDebug = printlnDebug;
   }
 
+  /**
+   * Tests must specify the timeout for expected outcome of asynchronous
+   * interactions. Longer timeout does not invalidate the correctness of
+   * the implementation, but can in some cases result in longer time to
+   * run the tests.
+   *
+   * @param defaultTimeoutMillis default timeout to be used in all expect* methods
+   * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore
+   */
+  public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis) {
+    this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, false);
+  }
+
   /**
    * Tests must specify the timeout for expected outcome of asynchronous
    * interactions. Longer timeout does not invalidate the correctness of
@@ -51,7 +67,7 @@ public TestEnvironment(long defaultTimeoutMillis, boolean printlnDebug) {
    * @param defaultTimeoutMillis default timeout to be used in all expect* methods
    */
   public TestEnvironment(long defaultTimeoutMillis) {
-    this(defaultTimeoutMillis, false);
+    this(defaultTimeoutMillis, defaultTimeoutMillis, false);
   }
 
   /**
@@ -67,7 +83,7 @@ public TestEnvironment(long defaultTimeoutMillis) {
    *                     often helpful to pinpoint simple race conditions etc.
    */
   public TestEnvironment(boolean printlnDebug) {
-    this(envDefaultTimeoutMillis(), printlnDebug);
+    this(envDefaultTimeoutMillis(), envDefaultNoSignalsTimeoutMillis(), printlnDebug);
   }
 
   /**
@@ -80,13 +96,22 @@ public TestEnvironment(boolean printlnDebug) {
    * or the default value ({@link TestEnvironment#DEFAULT_TIMEOUT_MILLIS}) will be used.
    */
   public TestEnvironment() {
-    this(envDefaultTimeoutMillis());
+    this(envDefaultTimeoutMillis(), envDefaultNoSignalsTimeoutMillis());
   }
 
+  /** This timeout is used when waiting for a signal to arrive. */
   public long defaultTimeoutMillis() {
     return defaultTimeoutMillis;
   }
 
+  /**
+   * This timeout is used when asserting that no further signals are emitted.
+   * Note that this timeout default
+   */
+  public long defaultNoSignalsTimeoutMillis() {
+    return defaultNoSignalsTimeoutMillis;
+  }
+
   /**
    * Tries to parse the env variable {@code DEFAULT_TIMEOUT_MILLIS} as long and returns the value if present OR its default value.
    *
@@ -97,11 +122,26 @@ public static long envDefaultTimeoutMillis() {
     if (envMillis == null) return DEFAULT_TIMEOUT_MILLIS;
     else try {
       return Long.parseLong(envMillis);
-    } catch(NumberFormatException ex) {
+    } catch (NumberFormatException ex) {
       throw new IllegalArgumentException(String.format("Unable to parse %s env value [%s] as long!", DEFAULT_TIMEOUT_MILLIS_ENV, envMillis), ex);
     }
   }
 
+  /**
+   * Tries to parse the env variable {@code DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS} as long and returns the value if present OR its default value.
+   *
+   * @throws java.lang.IllegalArgumentException when unable to parse the env variable
+   */
+  public static long envDefaultNoSignalsTimeoutMillis() {
+    final String envMillis = System.getenv(DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS_ENV);
+    if (envMillis == null) return envDefaultTimeoutMillis();
+    else try {
+      return Long.parseLong(envMillis);
+    } catch (NumberFormatException ex) {
+      throw new IllegalArgumentException(String.format("Unable to parse %s env value [%s] as long!", DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS_ENV, envMillis), ex);
+    }
+  }
+
   /**
    * To flop means to "fail asynchronously", either by onErroring or by failing some TCK check triggered asynchronously.
    * This method does *NOT* fail the test - it's up to inspections of the error to fail the test if required.
@@ -227,7 +267,7 @@ public Throwable dropAsyncError() {
    * were signalled pior to, or during that time (by calling {@code flop()}).
    */
   public void verifyNoAsyncErrors() {
-    verifyNoAsyncErrors(defaultTimeoutMillis());
+    verifyNoAsyncErrors(defaultNoSignalsTimeoutMillis());
   }
 
   /**
@@ -485,11 +525,11 @@ public <E extends Throwable> E expectError(Class<E> expected, long timeoutMillis
     }
 
     public void expectNone() throws InterruptedException {
-      expectNone(env.defaultTimeoutMillis());
+      expectNone(env.defaultNoSignalsTimeoutMillis());
     }
 
     public void expectNone(String errMsgPrefix) throws InterruptedException {
-      expectNone(env.defaultTimeoutMillis(), errMsgPrefix);
+      expectNone(env.defaultNoSignalsTimeoutMillis(), errMsgPrefix);
     }
 
     public void expectNone(long withinMillis) throws InterruptedException {

tck/src/test/java/org/reactivestreams/tck/IdentityProcessorVerificationTest.java

@@ -19,6 +19,7 @@
 public class IdentityProcessorVerificationTest extends TCKVerificationSupport {
 
   static final long DEFAULT_TIMEOUT_MILLIS = TestEnvironment.envDefaultTimeoutMillis();
+  static final long DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS = TestEnvironment.envDefaultNoSignalsTimeoutMillis();
 
   private ExecutorService ex;
   @BeforeClass void before() { ex = Executors.newFixedThreadPool(4); }
@@ -155,7 +156,7 @@ static class NoopProcessor implements Processor<Integer, Integer> {
   }
 
   private TestEnvironment newTestEnvironment() {
-    return new TestEnvironment(DEFAULT_TIMEOUT_MILLIS);
+    return new TestEnvironment(DEFAULT_TIMEOUT_MILLIS, DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS);
   }