TCK: Receptacle#expectError timeout approach

Motivation:
`defaultTimeoutMillis` is typically used as a "max time to wait until an expected event occurs". For example this value is used as the argument to a `ArrayBlockingQueue#poll(timeout)` [1][2][3] method call and therefore when things work as expected you rarely wait this full duration. However [Receptacle#expectError](https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java#L1030) does a `Thread.sleep(..)` and then checks if there is something present in the error list (which is a [CopyOnWriteArrayList](https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java#L45)). The value for the sleep is typically derived from [defaultTimeoutMillis](https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java#L521-L526) which leads to unconditional delays. This makes it challenging to use `defaultTimeoutMillis` in tests to specify "maximum amount of time to wait until some event occurs". The impacts are if `defaultTimeoutMillis` is set very small you are more likely to see timeouts (especially on over utilized CI servers), but if `defaultTimeoutMillis` the tests take a long time to complete.

[1] https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java#L1019
[2] https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java#L978
[3] https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java#L990

Modifications:
- TestEnvironment now supports an additional poll timeout parameter to be used
  when there is no explicit async event to trigger off of.

Result:
TestEnvironment's defaultTimeoutMillis can be used as "max time until some event
occurs" where defaultPollTimeoutMillis provides the polling interval to check
for an event if there is no explicit async event to trigger off of.
Fixes reactive-streams#451

Author: Scottmitch

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

@@ -14,8 +14,8 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-import org.reactivestreams.tck.flow.support.SubscriberBufferOverflowException;
 import org.reactivestreams.tck.flow.support.Optional;
+import org.reactivestreams.tck.flow.support.SubscriberBufferOverflowException;
 
 import java.util.Collections;
 import java.util.LinkedList;
@@ -24,7 +24,6 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.testng.Assert.assertTrue;
@@ -39,6 +38,7 @@ public class TestEnvironment {
   private static final String DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS_ENV = "DEFAULT_NO_SIGNALS_TIMEOUT_MILLIS";
 
   private final long defaultTimeoutMillis;
+  private final long defaultPollTimeoutMillis;
   private final long defaultNoSignalsTimeoutMillis;
   private final boolean printlnDebug;
 
@@ -50,15 +50,47 @@ public class TestEnvironment {
    * 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 defaultPollTimeoutMillis default amount of time to poll for events if {@code defaultTimeoutMillis} isn't
+   *                                 preempted by an asynchronous event.
    * @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,
    */
-  public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, boolean printlnDebug) {
+  public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, long defaultPollTimeoutMillis,
+                         boolean printlnDebug) {
     this.defaultTimeoutMillis = defaultTimeoutMillis;
+    this.defaultPollTimeoutMillis = defaultPollTimeoutMillis;
     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
+   * @param printlnDebug         if true, signals such as OnNext / Request / OnComplete etc will be printed to standard output,
+   */
+  public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, boolean printlnDebug) {
+    this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, defaultTimeoutMillis, 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
+   * @param defaultPollTimeoutMillis default amount of time to poll for events if {@code defaultTimeoutMillis} isn't
+   *                                 preempted by an asynchronous event.
+   */
+  public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis, long defaultPollTimeoutMillis) {
+      this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, defaultPollTimeoutMillis, false);
+  }
+
   /**
    * Tests must specify the timeout for expected outcome of asynchronous
    * interactions. Longer timeout does not invalidate the correctness of
@@ -69,7 +101,7 @@ public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMi
    * @param defaultNoSignalsTimeoutMillis default timeout to be used when no further signals are expected anymore
    */
   public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMillis) {
-    this(defaultTimeoutMillis, defaultNoSignalsTimeoutMillis, false);
+    this(defaultTimeoutMillis, defaultTimeoutMillis, defaultNoSignalsTimeoutMillis);
   }
 
   /**
@@ -81,7 +113,7 @@ public TestEnvironment(long defaultTimeoutMillis, long defaultNoSignalsTimeoutMi
    * @param defaultTimeoutMillis default timeout to be used in all expect* methods
    */
   public TestEnvironment(long defaultTimeoutMillis) {
-    this(defaultTimeoutMillis, defaultTimeoutMillis, false);
+    this(defaultTimeoutMillis, defaultTimeoutMillis, defaultTimeoutMillis);
   }
 
   /**
@@ -519,42 +551,57 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru
     }
 
     public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, String requiredMessagePart) throws Exception {
-      expectErrorWithMessage(expected, requiredMessagePart, env.defaultTimeoutMillis());
+      expectErrorWithMessage(expected, Collections.singletonList(requiredMessagePart), env.defaultTimeoutMillis(), env.defaultPollTimeoutMillis);
     }
     public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, List<String> requiredMessagePartAlternatives) throws Exception {
-      expectErrorWithMessage(expected, requiredMessagePartAlternatives, env.defaultTimeoutMillis());
+      expectErrorWithMessage(expected, requiredMessagePartAlternatives, env.defaultTimeoutMillis(), env.defaultPollTimeoutMillis);
     }
 
     @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
     public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, String requiredMessagePart, long timeoutMillis) throws Exception {
       expectErrorWithMessage(expected, Collections.singletonList(requiredMessagePart), timeoutMillis);
     }
+
     public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, List<String> requiredMessagePartAlternatives, long timeoutMillis) throws Exception {
-      final E err = expectError(expected, timeoutMillis);
+      expectErrorWithMessage(expected, requiredMessagePartAlternatives, timeoutMillis, timeoutMillis);
+    }
+
+    public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, List<String> requiredMessagePartAlternatives,
+                                                             long totalTimeoutMillis, long pollTimeoutMillis) throws Exception {
+      final E err = expectError(expected, totalTimeoutMillis, pollTimeoutMillis);
       final String message = err.getMessage();
-      
+
       boolean contains = false;
-      for (String requiredMessagePart : requiredMessagePartAlternatives) 
+      for (String requiredMessagePart : requiredMessagePartAlternatives)
         if (message.contains(requiredMessagePart)) contains = true; // not short-circuting loop, it is expected to
       assertTrue(contains,
-                 String.format("Got expected exception [%s] but missing message part [%s], was: %s",
-                               err.getClass(), "anyOf: " + requiredMessagePartAlternatives, err.getMessage()));
+              String.format("Got expected exception [%s] but missing message part [%s], was: %s",
+                      err.getClass(), "anyOf: " + requiredMessagePartAlternatives, err.getMessage()));
     }
 
     public <E extends Throwable> E expectError(Class<E> expected) throws Exception {
       return expectError(expected, env.defaultTimeoutMillis());
     }
 
     public <E extends Throwable> E expectError(Class<E> expected, long timeoutMillis) throws Exception {
-      return expectError(expected, timeoutMillis, String.format("Expected onError(%s)", expected.getName()));
+      return expectError(expected, timeoutMillis, env.defaultPollTimeoutMillis);
     }
 
     public <E extends Throwable> E expectError(Class<E> expected, String errorMsg) throws Exception {
       return expectError(expected, env.defaultTimeoutMillis(), errorMsg);
     }
 
     public <E extends Throwable> E expectError(Class<E> expected, long timeoutMillis, String errorMsg) throws Exception {
-      return received.expectError(expected, timeoutMillis, errorMsg);
+      return expectError(expected, timeoutMillis, env.defaultPollTimeoutMillis, errorMsg);
+    }
+
+    public <E extends Throwable> E expectError(Class<E> expected, long totalTimeoutMillis, long pollTimeoutMillis) throws Exception {
+      return expectError(expected, totalTimeoutMillis, pollTimeoutMillis, String.format("Expected onError(%s)", expected.getName()));
+    }
+
+    public <E extends Throwable> E expectError(Class<E> expected, long totalTimeoutMillis, long pollTimeoutMillis,
+                                               String errorMsg) throws Exception {
+      return received.expectError(expected, totalTimeoutMillis, pollTimeoutMillis, errorMsg);
     }
 
     public void expectNone() throws InterruptedException {
@@ -1025,22 +1072,44 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru
       } // else, ok
     }
 
-    @SuppressWarnings("unchecked")
+    /**
+     * @deprecated Deprecated in favor of {@link #expectError(Class, long, long, String)}.
+     */
+    @Deprecated
     public <E extends Throwable> E expectError(Class<E> clazz, long timeoutMillis, String errorMsg) throws Exception {
-      Thread.sleep(timeoutMillis);
-
-      if (env.asyncErrors.isEmpty()) {
-        return env.flopAndFail(String.format("%s within %d ms", errorMsg, timeoutMillis));
-      } else {
-        // ok, there was an expected error
-        Throwable thrown = env.asyncErrors.remove(0);
+      return expectError(clazz, timeoutMillis, timeoutMillis, errorMsg);
+    }
 
-        if (clazz.isInstance(thrown)) {
-          return (E) thrown;
+    @SuppressWarnings("unchecked")
+    final <E extends Throwable> E expectError(Class<E> clazz, final long totalTimeoutMillis,
+                                              long pollTimeoutMillis,
+                                              String errorMsg) throws Exception {
+      long totalTimeoutRemainingMillis = totalTimeoutMillis;
+      long timeStampA = System.nanoTime();
+      long timeStampB;
+
+      for (;;) {
+        Thread.sleep(Math.min(pollTimeoutMillis, totalTimeoutRemainingMillis));
+
+        if (env.asyncErrors.isEmpty()) {
+          timeStampB = System.nanoTime();
+          totalTimeoutRemainingMillis =- timeStampB - timeStampA;
+          timeStampA = timeStampB;
+
+          if (totalTimeoutRemainingMillis <= 0) {
+            return env.flopAndFail(String.format("%s within %d ms", errorMsg, totalTimeoutMillis));
+          }
         } else {
+          // ok, there was an expected error
+          Throwable thrown = env.asyncErrors.remove(0);
 
-          return env.flopAndFail(String.format("%s within %d ms; Got %s but expected %s",
-                                               errorMsg, timeoutMillis, thrown.getClass().getCanonicalName(), clazz.getCanonicalName()));
+          if (clazz.isInstance(thrown)) {
+            return (E) thrown;
+          } else {
+
+            return env.flopAndFail(String.format("%s within %d ms; Got %s but expected %s",
+                    errorMsg, totalTimeoutMillis, thrown.getClass().getCanonicalName(), clazz.getCanonicalName()));
+          }
         }
       }
     }