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

akarnokd · 2017-04-03T14:15:10Z

This PR extends the TCK to support two features of Processor implementations:

Replay all

Some Processor implementations may replay all onNext items they received over their lifetime to the incoming Subscriber. The required_mustRequestFromUpstreamForElementsThatHaveBeenRequestedLongAgo however expected that a late Subscriber receives only the very latest onNext item and fail the test if any of the first two was observed. The enhancement allows observing (x, y, z) or (y, z) or (z).

Note that indeed this change may let Processor implementations pass which were supposed to emit only z. My opinion is that such implementation should already have unit tests verifying its emission pattern and the TCK should verify the implementation honors the RS protocol, even if it means getting 1-3 onNexts before the terminal event. If this is a problem, the change could be gated behind a IdentityProcessorVerification overridable property method, similar to the next change.

Emit all onNexts before an onError

Some Processor implementations may relay/replay all received onNexts items before they relay/replay a received onError event. This allows valid items to be processed before the sequence terminates with an error. Since this behavior may be specified explicitly on a Processor implementation, the TCK's IdentityProcessorVerification has been extended with a strictEventOrdering() method (default false) that when true, the otherwise failing required_spec104_mustCallOnErrorOnAllItsSubscribersIfItEncountersANonRecoverableError first consumes the pending item and then verifies the error state.

Since this change requires overriding a method, existing Processor implementations that eagerly call onError should still pass or fail this test as before.

ReplayProcessor

To demonstrate a Processor implementation that can exhibit both behaviors, a custom and package-private ReplayProcessor implementation has been added along with its TCK tests for non-strict and strict mode. (Note that this particular processor cancels its upstream if all downstream Subscribers have cancelled before the upstream's terminal event. This behavior is uncommon as ReactiveX style Processors keep running and relay new events to later Subscribers after the first set is gone.)

Related: reactive-streams/reactive-streams-dotnet#30

viktorklang · 2017-04-04T08:25:12Z

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

@@ -738,6 +774,20 @@ public void expectNextElement(ManualSubscriber<T> sub, T expected) throws Interr
      }
    }

+    public void expectAnyNextElement(ManualSubscriber<T> sub, Collection<T> expected) throws InterruptedException {


Shouldn't expected be a Set?

Set extends Collection, plus there is no Set.of in pre Java-9 but there is Arrays.asList(). For the particular test, there are 3 elements to expect at most so asList is okay, for any other use, you can call it with a Set if you want to.

Since this is an API I'd rather have the types be specific.

private final <T extends Comparable<T>> Set<T> toSet(T... ts) { final Set<T> set; if (ts.length == 0) set = Collections.emptySet(); else { set = new HashSet<T>(); for(final T t : ts) set.add(t); } return set; }

Indeed, this is an API that should be as enabling as possible, that means using the most general type from the type hierarchy that has a contains method.

@akarnokd I disagree, the semantics of expected is that of a Set, which is why Set is the most general type with the correct semantics, instead of requiring a HashSet or a TreeSet etc.

viktorklang · 2017-04-04T08:36:47Z

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

@@ -973,6 +977,21 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru
      } // else, ok
    }

+    public boolean tryExpectCompletion(long timeoutMillis) throws InterruptedException {
+      while (timeoutMillis-- > 0) {


This is going to be wrong as it assumes it'll take a millisecond per pass.

viktorklang · 2017-04-04T08:36:56Z

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

+          }
+          return false;
+        }
+        Thread.sleep(1);


Why this ISO poll(remaining)?

I don't understand this question. This routine uses peek because in case the next received element is not a terminal indicator, that element should remain available for any of the other next calls that may want to verify the item.

@akarnokd if you use abq.poll(remaining) instead of abq.peek() + abp.poll() you do not need to do the sleep(1)-dance.

There is no such method of abq.poll(arg0): Javadoc

The purpose of the method is to consume a non-defined, non-null Optional and return true indicating the next event in line was a terminal event. If the next event is not a terminal event, don't consume it and return false and leave up to the test what to do with that situation. In other terms, poll() is not good because it consumes the item unconditionally.

1: https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ArrayBlockingQueue.html#poll(long,%20java.util.concurrent.TimeUnit)
2: Good point. Seems like the API is missing a peek(timeout, unit)!

akarnokd · 2017-04-04T09:17:18Z

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

@@ -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 {


viktorklang · 2017-05-22T08:02:00Z

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

+   * 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() {


I'm n ot super-convinced by the name. Isn't this more about eagerTermination?

or shortCircuitTermination?

viktorklang · 2017-05-22T08:06:52Z

@reactive-streams/contributors Yay or nay?

akarnokd · 2017-12-22T14:11:52Z

I think such non-identity processor could have separate standalone TCK class instead.

Enable replay-like and delay-error like Processor impls

e1cda5f

akarnokd mentioned this pull request Apr 3, 2017

RC1 TCK: IdentityProcessorVerification expectations reactive-streams/reactive-streams-dotnet#30

Open

akarnokd changed the title ~~Enable replay-like and delay-error like Processor impls~~ TCK: Enable replay-like and delay-error like Processor impls Apr 3, 2017

Don't consume a non-terminal signal

3202772

viktorklang reviewed Apr 4, 2017

View reviewed changes

akarnokd added 2 commits April 4, 2017 10:45

Don't assume sleep(1) actually sleeps 1 millisecond

c2c54bc

Use Set with expectAnyNextElement()

eddb80f

akarnokd commented Apr 4, 2017

View reviewed changes

viktorklang reviewed May 22, 2017

View reviewed changes

akarnokd closed this Dec 22, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

akarnokd commented Apr 3, 2017 •

edited

Loading

viktorklang Apr 4, 2017

akarnokd Apr 4, 2017

viktorklang Apr 4, 2017

akarnokd Apr 4, 2017

viktorklang Apr 4, 2017

viktorklang Apr 4, 2017

akarnokd Apr 4, 2017

viktorklang Apr 4, 2017

akarnokd Apr 4, 2017

viktorklang Apr 4, 2017

akarnokd Apr 4, 2017

viktorklang Apr 4, 2017

akarnokd Apr 4, 2017

viktorklang May 22, 2017

viktorklang May 22, 2017

viktorklang commented May 22, 2017

akarnokd commented Dec 22, 2017

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

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

Conversation

akarnokd commented Apr 3, 2017 • edited Loading

Replay all

Emit all onNexts before an onError

ReplayProcessor

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viktorklang commented May 22, 2017

akarnokd commented Dec 22, 2017

akarnokd commented Apr 3, 2017 •

edited

Loading