+tck timeouts now have default values as well as are ENV overrideable

Author: ktoso

.travis.yml

@@ -9,3 +9,5 @@ jdk:
 env:
   global:
   - TERM=dumb
+  - DEFAULT_TIMEOUT_MILLIS=300
+  - PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=300

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

@@ -17,14 +17,13 @@
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class AsyncSubscriberTest extends SubscriberBlackboxVerification<Integer> {
-  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(DEFAULT_TIMEOUT_MILLIS));
+    super(new TestEnvironment());
   }
 
   @Override public Subscriber<Integer> createSubscriber() {
@@ -53,7 +52,7 @@ public AsyncSubscriberTest() {
     };
 
     new NumberIterablePublisher(0, 10, e).subscribe(sub);
-    latch.await(DEFAULT_TIMEOUT_MILLIS * 10, TimeUnit.MILLISECONDS);
+    latch.await(env.defaultTimeoutMillis() * 10, TimeUnit.MILLISECONDS);
     assertEquals(i.get(), 45);
   }
 

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

@@ -15,15 +15,12 @@
 @Test // Must be here for TestNG to find and run this, do not remove
 public class IterablePublisherTest extends PublisherVerification<Integer> {
 
-  final static long DefaultTimeoutMillis = 100;
-  final static long PublisherReferenceGCTimeoutMillis = 300;
-
   private ExecutorService e;
   @BeforeClass void before() { e = Executors.newFixedThreadPool(4); }
   @AfterClass void after() { if (e != null) e.shutdown(); }
 
   public IterablePublisherTest() {
-    super(new TestEnvironment(DefaultTimeoutMillis), PublisherReferenceGCTimeoutMillis);
+    super(new TestEnvironment());
   }
 
   @SuppressWarnings("unchecked")

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

@@ -13,14 +13,12 @@
 @Test // Must be here for TestNG to find and run this, do not remove
 public class SyncSubscriberTest extends SubscriberBlackboxVerification<Integer> {
 
-  final static long DefaultTimeoutMillis = 100;
-
   private ExecutorService e;
   @BeforeClass void before() { e = Executors.newFixedThreadPool(4); }
   @AfterClass void after() { if (e != null) e.shutdown(); }
 
   public SyncSubscriberTest() {
-    super(new TestEnvironment(DefaultTimeoutMillis));
+    super(new TestEnvironment());
   }
 
   @Override public Subscriber<Integer> createSubscriber() {

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

@@ -11,15 +11,13 @@
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class UnboundedIntegerIncrementPublisherTest extends PublisherVerification<Integer> {
-  final static long DefaultTimeoutMillis = 200;
-  final static long PublisherReferenceGCTimeoutMillis = 500;
 
   private ExecutorService e;
   @BeforeClass void before() { e = Executors.newFixedThreadPool(4); }
   @AfterClass void after() { if (e != null) e.shutdown(); }
 
   public UnboundedIntegerIncrementPublisherTest() {
-    super(new TestEnvironment(DefaultTimeoutMillis), PublisherReferenceGCTimeoutMillis);
+    super(new TestEnvironment());
   }
 
   @Override public Publisher<Integer> createPublisher(long elements) {

tck/README.md

@@ -132,11 +132,8 @@ import org.reactivestreams.tck.TestEnvironment;
 
 public class RangePublisherTest extends PublisherVerification<Integer> {
 
-  public static final long DEFAULT_TIMEOUT_MILLIS = 300L;
-  public static final long PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 1000L;
-
   public RangePublisherTest() {
-    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS);
+    super(new TestEnvironment());
   }
 
   @Override
@@ -174,6 +171,39 @@ Notable configuration options include:
 * `boundedDepthOfOnNextAndRequestRecursion` – which should only be overridden in case of synchronous Publishers. This number will be used to validate if a
 `Subscription` actually solves the "unbounded recursion" problem (Rule 3.3).
 
+### Timeout configuration
+Publisher tests make use of two kinds of timeouts, one is the `defaultTimeoutMillis` which corresponds to all methods used
+within the TCK which await for something to happen. The other timeout is `publisherReferenceGCTimeoutMillis` which is only used in order to verify
+[Rule 3.13](https://github.com/reactive-streams/reactive-streams#3.13) which defines that subscriber references MUST be dropped
+by the Publisher.
+
+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 just:
+
+```bash
+export DEFAULT_TIMEOUT_MILLIS=300
+export PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS=500
+```
+
+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 PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS = 500L;
+
+  public RangePublisherTest() {
+    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS), PUBLISHER_REFERENCE_CLEANUP_TIMEOUT_MILLIS);
+  }
+
+  // ...
+}
+```
+
+Note that hard-coded values *take precedence* over environment set values (!).
+
 ## Subscriber Verification
 
 Subscriber rules Verification is split up into two files (styles) of tests.
@@ -233,10 +263,8 @@ import org.reactivestreams.tck.TestEnvironment;
 
 public class MySubscriberBlackboxVerificationTest extends SubscriberBlackboxVerification<Integer> {
 
-  public static final long DEFAULT_TIMEOUT_MILLIS = 300L;
-
   public MySubscriberBlackboxVerificationTest() {
-    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS));
+    super(new TestEnvironment());
   }
 
   @Override
@@ -273,10 +301,8 @@ import org.reactivestreams.tck.TestEnvironment;
 
 public class MySubscriberWhiteboxVerificationTest extends SubscriberWhiteboxVerification<Integer> {
 
-  public static final long DEFAULT_TIMEOUT_MILLIS = 300L;
-
   public MySubscriberWhiteboxVerificationTest() {
-    super(new TestEnvironment(DEFAULT_TIMEOUT_MILLIS));
+    super(new TestEnvironment());
   }
 
   @Override
@@ -331,6 +357,34 @@ public class MySubscriberWhiteboxVerificationTest extends SubscriberWhiteboxVeri
 }
 ```
 
+### Timeout configuration
+Similarily to `PublisherVerification`, it is possible to set the timeouts used by the TCK to validate subscriber behaviour.
+This can be set either by using env variables or hardcoded explicitly.
+
+**Use env variables** to set the timeout value to be used by the TCK:
+
+```bash
+export DEFAULT_TIMEOUT_MILLIS=300
+```
+
+Or **define the timeout explicitly in code**:
+
+```java
+public class MySubscriberTest extends BlackboxSubscriberVerification<Integer> {
+
+  public static final long DEFAULT_TIMEOUT_MILLIS = 300L;
+
+  public RangePublisherTest() {
+    super(new MySubscriberTest(DEFAULT_TIMEOUT_MILLIS));
+  }
+
+  // ...
+}
+```
+
+Note that hard-coded values *take precedence* over environment set values (!).
+
+
 ## Subscription Verification
 
 Please note that while `Subscription` does **not** have it's own test class, it's rules are validated inside of the `Publisher` and `Subscriber` tests – depending if the Rule demands specific action to be taken by the publishing, or subscribing side of the subscription contract.

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

@@ -36,27 +36,38 @@ public abstract class IdentityProcessorVerification<T> extends WithHelperPublish
   // without dropping elements. Defaults to `TestEnvironment.TEST_BUFFER_SIZE`.
   private final int processorBufferSize;
 
+  /**
+   * Test class must specify the expected time it takes for the publisher to
+   * shut itself down when the the last downstream {@code Subscription} is cancelled.
+   *
+   * The processor will be required to be able to buffer {@code TestEnvironment.TEST_BUFFER_SIZE} elements.
+   */
+  @SuppressWarnings("unused")
+  public IdentityProcessorVerification(final TestEnvironment env) {
+    this(env, PublisherVerification.envPublisherReferenceGCTimeoutMillis(), TestEnvironment.TEST_BUFFER_SIZE);
+  }
+
   /**
    * Test class must specify the expected time it takes for the publisher to
    * shut itself down when the the last downstream {@code Subscription} is cancelled.
    *
    * The processor will be required to be able to buffer {@code TestEnvironment.TEST_BUFFER_SIZE} elements.
    *
-   * @param publisherShutdownTimeoutMillis expected time which a processor requires to shut itself down
+   * @param publisherReferenceGCTimeoutMillis used to determine after how much time a reference to a Subscriber should be already dropped by the Publisher.
    */
   @SuppressWarnings("unused")
-  public IdentityProcessorVerification(final TestEnvironment env, long publisherShutdownTimeoutMillis) {
-    this(env, publisherShutdownTimeoutMillis, TestEnvironment.TEST_BUFFER_SIZE);
+  public IdentityProcessorVerification(final TestEnvironment env, long publisherReferenceGCTimeoutMillis) {
+    this(env, publisherReferenceGCTimeoutMillis, TestEnvironment.TEST_BUFFER_SIZE);
   }
 
   /**
    * Test class must specify the expected time it takes for the publisher to
    * shut itself down when the the last downstream {@code Subscription} is cancelled.
    *
-   * @param publisherShutdownTimeoutMillis expected time which a processor requires to shut itself down
+   * @param publisherReferenceGCTimeoutMillis used to determine after how much time a reference to a Subscriber should be already dropped by the Publisher.
    * @param processorBufferSize            number of elements the processor is required to be able to buffer.
    */
-  public IdentityProcessorVerification(final TestEnvironment env, long publisherShutdownTimeoutMillis, int processorBufferSize) {
+  public IdentityProcessorVerification(final TestEnvironment env, long publisherReferenceGCTimeoutMillis, int processorBufferSize) {
     this.env = env;
     this.processorBufferSize = processorBufferSize;
 
@@ -76,7 +87,7 @@ public Publisher<T> createHelperPublisher(long elements) {
       }
     };
 
-    publisherVerification = new PublisherVerification<T>(env, publisherShutdownTimeoutMillis) {
+    publisherVerification = new PublisherVerification<T>(env, publisherReferenceGCTimeoutMillis) {
       @Override
       public Publisher<T> createPublisher(long elements) {
         return IdentityProcessorVerification.this.createPublisher(elements);

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

@@ -33,14 +33,50 @@
  */
 public abstract class PublisherVerification<T> implements PublisherVerificationRules {
 
+  private static final String PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS_ENV = "PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS";
+  private static final long DEFAULT_PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS = 300L;
+
   private final TestEnvironment env;
   private final long publisherReferenceGCTimeoutMillis;
 
+  /**
+   * Constructs a new verification class using the given env and configuration.
+   *
+   * @param publisherReferenceGCTimeoutMillis used to determine after how much time a reference to a Subscriber should be already dropped by the Publisher.
+   */
   public PublisherVerification(TestEnvironment env, long publisherReferenceGCTimeoutMillis) {
     this.env = env;
     this.publisherReferenceGCTimeoutMillis = publisherReferenceGCTimeoutMillis;
   }
 
+  /**
+   * Constructs a new verification class using the given env and configuration.
+   *
+   * The value for {@code publisherReferenceGCTimeoutMillis} will be obtained by using {@link PublisherVerification#envPublisherReferenceGCTimeoutMillis()}.
+   */
+  public PublisherVerification(TestEnvironment env) {
+    this.env = env;
+    this.publisherReferenceGCTimeoutMillis = envPublisherReferenceGCTimeoutMillis();
+  }
+
+  /**
+   * Tries to parse the env variable {@code PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS} as long and returns the value if present,
+   * OR its default value ({@link PublisherVerification#DEFAULT_PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS}).
+   *
+   * This value is used to determine after how much time a reference to a Subscriber should be already dropped by the Publisher.
+   *
+   * @throws java.lang.IllegalArgumentException when unable to parse the env variable
+   */
+  public static long envPublisherReferenceGCTimeoutMillis() {
+    final String envMillis = System.getenv(PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS_ENV);
+    if (envMillis == null) return DEFAULT_PUBLISHER_REFERENCE_GC_TIMEOUT_MILLIS;
+    else try {
+      return Long.parseLong(envMillis);
+    } 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);
+    }
+  }
+
   /**
    * This is the main method you must implement in your test incarnation.
    * It must create a Publisher for a stream with exactly the given number of elements.
@@ -88,6 +124,14 @@ public long boundedDepthOfOnNextAndRequestRecursion() {
     return 1;
   }
 
+  /**
+   * The amount of time after which a cancelled Subscriber reference should be dropped.
+   * See Rule 3.13 for details.
+   */
+  final public long publisherReferenceGCTimeoutMillis() {
+    return publisherReferenceGCTimeoutMillis;
+  }
+
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
 
   @BeforeMethod
@@ -298,20 +342,20 @@ public void onComplete() {
   public void optional_spec104_mustSignalOnErrorWhenFails() throws Throwable {
     try {
       whenHasErrorPublisherTest(new PublisherTestRun<T>() {
-          @Override
-          public void run(final Publisher<T> pub) throws InterruptedException {
-              final Latch latch = new Latch(env);
-              pub.subscribe(new TestEnvironment.TestSubscriber<T>(env) {
-                  @Override
-                  public void onError(Throwable cause) {
-                      latch.assertOpen(String.format("Error-state Publisher %s called `onError` twice on new Subscriber", pub));
-                      latch.close();
-                  }
-              });
+        @Override
+        public void run(final Publisher<T> pub) throws InterruptedException {
+          final Latch latch = new Latch(env);
+          pub.subscribe(new TestEnvironment.TestSubscriber<T>(env) {
+            @Override
+            public void onError(Throwable cause) {
+              latch.assertOpen(String.format("Error-state Publisher %s called `onError` twice on new Subscriber", pub));
+              latch.close();
+            }
+          });
 
-              latch.expectClose(String.format("Error-state Publisher %s did not call `onError` on new Subscriber", pub));
+          latch.expectClose(String.format("Error-state Publisher %s did not call `onError` on new Subscriber", pub));
 
-              env.verifyNoAsyncErrors(env.defaultTimeoutMillis());
+          env.verifyNoAsyncErrors(env.defaultTimeoutMillis());
           }
       });
     } catch (SkipException se) {
@@ -404,25 +448,25 @@ public void run(Publisher<T> pub) throws Throwable {
   @Override @Test
   public void required_spec112_mayRejectCallsToSubscribeIfPublisherIsUnableOrUnwillingToServeThemRejectionMustTriggerOnErrorInsteadOfOnSubscribe() throws Throwable {
     whenHasErrorPublisherTest(new PublisherTestRun<T>() {
-        @Override
-        public void run(Publisher<T> pub) throws Throwable {
-            final Latch onErrorLatch = new Latch(env);
-            ManualSubscriberWithSubscriptionSupport<T> sub = new ManualSubscriberWithSubscriptionSupport<T>(env) {
-                @Override
-                public void onError(Throwable cause) {
-                    onErrorLatch.assertOpen("Only one onError call expected");
-                    onErrorLatch.close();
-                }
+      @Override
+      public void run(Publisher<T> pub) throws Throwable {
+        final Latch onErrorLatch = new Latch(env);
+        ManualSubscriberWithSubscriptionSupport<T> sub = new ManualSubscriberWithSubscriptionSupport<T>(env) {
+          @Override
+          public void onError(Throwable cause) {
+            onErrorLatch.assertOpen("Only one onError call expected");
+            onErrorLatch.close();
+          }
 
-                @Override
-                public void onSubscribe(Subscription subs) {
-                    env.flop("onSubscribe should not be called if Publisher is unable to subscribe a Subscriber");
-                }
-            };
-            pub.subscribe(sub);
-            onErrorLatch.assertClosed("Should have received onError");
+          @Override
+          public void onSubscribe(Subscription subs) {
+            env.flop("onSubscribe should not be called if Publisher is unable to subscribe a Subscriber");
+          }
+        };
+        pub.subscribe(sub);
+        onErrorLatch.assertClosed("Should have received onError");
 
-            env.verifyNoAsyncErrors();
+        env.verifyNoAsyncErrors();
         }
     });
   }