Error message about doesCoordinatedEmission.

Author: akarnokd

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

@@ -450,6 +450,8 @@ public TestSetup apply(Long aLong) throws Throwable {
             sub1.request(1);
             sub2.request(1);
 
+            expectRequest();
+
             final T x = sendNextTFromUpstream();
 
             expectNextElement(sub1, x);
@@ -459,9 +461,18 @@ public TestSetup apply(Long aLong) throws Throwable {
             sub2.request(1);
           } else {
             sub1.request(1);
-            expectRequest();
+
+            expectRequest(env.defaultTimeoutMillis(),
+                    "If the Processor coordinates requests/emissions when having multiple Subscribers"
+                    + " at once, please override doesCoordinatedEmission() in this "
+                    + "IdentityProcessorVerification to allow this test to pass.");
+
             final T x = sendNextTFromUpstream();
-            expectNextElement(sub1, x);
+            expectNextElement(sub1, x,
+                    "If the Processor coordinates requests/emissions when having multiple Subscribers"
+                            + " at once, please override doesCoordinatedEmission() in this "
+                            + "IdentityProcessorVerification to allow this test to pass.");
+
             sub1.request(1);
 
             // sub1 has received one element, and has one demand pending
@@ -744,7 +755,10 @@ public TestSetup apply(Long subscribers) throws Throwable {
             expectNextElement(sub2, z);
           } else {
             final T z = sendNextTFromUpstream();
-            expectNextElement(sub1, z);
+            expectNextElement(sub1, z,
+                    "If the Processor coordinates requests/emissions when having multiple Subscribers"
+                            + " at once, please override doesCoordinatedEmission() in this "
+                            + "IdentityProcessorVerification to allow this test to pass.");
             sub2.expectNone(); // since sub2 hasn't requested anything yet
 
             sub2.request(1);
@@ -818,6 +832,13 @@ public void expectNextElement(ManualSubscriber<T> sub, T expected) throws Interr
       }
     }
 
+    public void expectNextElement(ManualSubscriber<T> sub, T expected, String errorMessageAddendum) throws InterruptedException {
+      final T elem = sub.nextElement(String.format("timeout while awaiting %s. %s", expected, errorMessageAddendum));
+      if (!elem.equals(expected)) {
+        env.flop(String.format("Received `onNext(%s)` on downstream but expected `onNext(%s)`", elem, expected));
+      }
+    }
+
     public T sendNextTFromUpstream() throws InterruptedException {
       final T x = nextT();
       sendNext(x);

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

@@ -763,6 +763,17 @@ public long expectRequest(long timeoutMillis) throws InterruptedException {
       }
     }
 
+
+    public long expectRequest(long timeoutMillis, String errorMessageAddendum) throws InterruptedException {
+      long requested = requests.next(timeoutMillis, String.format("Did not receive expected `request` call. %s", errorMessageAddendum));
+      if (requested <= 0) {
+        return env.<Long>flopAndFail(String.format("Requests cannot be zero or negative but received request(%s)", requested));
+      } else {
+        pendingDemand += requested;
+        return requested;
+      }
+    }
+
     public void expectExactRequest(long expected) throws InterruptedException {
       expectExactRequest(expected, env.defaultTimeoutMillis());
     }