TCK: Enable replay-like and delay-error like Processor impls

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

@@ -11,6 +11,8 @@
 
 package org.reactivestreams.tck;
 
+import java.util.*;
+
 import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -25,9 +27,6 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
-import java.util.HashSet;
-import java.util.Set;
-
 public abstract class IdentityProcessorVerification<T> extends WithHelperPublisher<T>
   implements SubscriberWhiteboxVerificationRules, PublisherVerificationRules {
 
@@ -189,6 +188,19 @@ public long maxSupportedSubscribers() {
       return Long.MAX_VALUE;
   }
 
+  /**
+   * Indicates that the tested implementation keeps a strict event ordering in respect to
+   * {@code onNext} and {@code onError}.
+   * Some {@code Processor} implementation may emit all {@code onNext}s received before emitting 
+   * any {@code onError}, similar to how {@code onComplete} is usually emitted after all
+   * previous {@code onNext} items have been emitted. The default implementation returns false,
+   * indicating that {@code onError} may cut ahead and get emitted even if there are
+   * {@code onNext} events ready for consumption (via {@code request()}) by the {@code Subscriber}.
+   */
+  public boolean strictEventOrdering() {
+      return false;
+  }
+
   ////////////////////// TEST ENV CLEANUP /////////////////////////////////////
 
   @BeforeMethod
@@ -417,6 +429,14 @@ public TestSetup apply(Long aLong) throws Throwable {
           final Exception ex = new RuntimeException("Test exception");
           sendError(ex);
           sub1.expectError(ex);
+
+          // some Processors may only emit the terminal error if
+          // all previously submitted onNext value has been
+          // consumed by the Subscriber
+          if (strictEventOrdering()) {
+              sub2.request(1);
+              expectNextElement(sub2, x);
+          }
           sub2.expectError(ex);
 
           env.verifyNoAsyncErrorsNoDelay();
@@ -668,7 +688,10 @@ public TestSetup apply(Long subscribers) throws Throwable {
           sub2.expectNone(); // since sub2 hasn't requested anything yet
 
           sub2.request(1);
-          expectNextElement(sub2, z);
+          // some processors could emit items from the beginning, some may
+          // cache for a limited time or count, therefore
+          // any of the first 3 items may appear after requesing one
+          expectAnyNextElement(sub2, new HashSet<T>(Arrays.asList(x, y, z)));
 
           if (totalRequests == 3) {
             expectRequest();
@@ -677,7 +700,20 @@ public TestSetup apply(Long subscribers) throws Throwable {
           // to avoid error messages during test harness shutdown
           sendCompletion();
           sub1.expectCompletion(env.defaultTimeoutMillis());
-          sub2.expectCompletion(env.defaultTimeoutMillis());
+
+          // sub2 may not complete because it still has pending y or z
+          if (!sub2.tryExpectCompletion(env.defaultTimeoutMillis())) {
+              sub2.request(1);
+              expectAnyNextElement(sub2, new HashSet<T>(Arrays.asList(y, z)));
+
+              // z may still be pending
+              if (!sub2.tryExpectCompletion(env.defaultTimeoutMillis())) {
+                  sub2.request(1);
+                  expectNextElement(sub2, z);
+                  // after z, it should get onComplete reasonably quickly
+                  sub2.expectCompletion(env.defaultTimeoutMillis());
+              }
+          }
 
           env.verifyNoAsyncErrorsNoDelay();
         }};
@@ -738,6 +774,20 @@ public void expectNextElement(ManualSubscriber<T> sub, T expected) throws Interr
       }
     }
 
+    public void expectAnyNextElement(ManualSubscriber<T> sub, Set<T> expected) throws InterruptedException {
+      final T elem = sub.nextElement(String.format("timeout while awaiting %s", expected));
+      if (!expected.contains(elem)) {
+        StringBuilder sb = new StringBuilder();
+        for (T t : expected) {
+            if (sb.length() > 0) {
+                sb.append(", ");
+            }
+            sb.append(String.format("`onNext(%s)`", t));
+        }
+        env.flop(String.format("Received `onNext(%s)` on downstream but expected any of %s", elem, sb));
+      }
+    }
+
     public T sendNextTFromUpstream() throws InterruptedException {
       final T x = nextT();
       sendNext(x);

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

@@ -506,6 +506,10 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru
       received.expectCompletion(timeoutMillis, errorMsg);
     }
 
+    public boolean tryExpectCompletion(long timeoutMillis) throws InterruptedException {
+      return received.tryExpectCompletion(timeoutMillis);
+    }
+
     public <E extends Throwable> void expectErrorWithMessage(Class<E> expected, String requiredMessagePart) throws Exception {
       expectErrorWithMessage(expected, requiredMessagePart, env.defaultTimeoutMillis());
     }
@@ -973,6 +977,22 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru
       } // else, ok
     }
 
+    public boolean tryExpectCompletion(long timeoutMillis) throws InterruptedException {
+      long end = System.currentTimeMillis() + timeoutMillis;
+      do {
+        Optional<T> value = abq.peek();
+        if (value != null) {
+          if (!value.isDefined()) {
+            abq.poll();
+            return true;
+          }
+          return false;
+        }
+        Thread.sleep(1);
+      } while (System.currentTimeMillis() < end);
+      return false;
+    }
+
     @SuppressWarnings("unchecked")
     public <E extends Throwable> E expectError(Class<E> clazz, long timeoutMillis, String errorMsg) throws Exception {
       Thread.sleep(timeoutMillis);