Subscriber whitebox verification tests demand

[jroper]

So firstly, there's no javadocs on SubscriberPuppet. Is triggerRequest meant to trigger a request synchronously? Due to #277, it currently must be triggered synchronously, otherwise some of the tests violate the spec - but that's probably a problem with those tests.

Another thing not explained is whether exactly elements must be triggered. I suspect many subscribers (including the one I'm implementing, netty) do not allow this level of control over them, since the number of elements requested is controlled by buffers, and it's only possible for the test to signal an event to say it's possible to write more elements, not how many elements. Modifying the logic such that something external can trigger a specific number of elements would undermine the existing logic there, meaning that the TCK isn't testing the real subscriber.

So, as it happens, not requesting the exact number of elements still works, the TCK makes no assertions about how many elements were actually requested. But because of this, the TCK itself violates the spec, since it may end up sending more elements than were requested. Consider a streams implementation that only supports requesting one element at a time (such as iteratees), the TCK triggers a request for 2 elements, the iteratee requests one, and then the TCK sends 2 elements, a violation of the spec. It's only due to implementation details in each subscriber that this works.

So I'd like to propose that these tests be modified in the following way:

SubscriberPuppet.triggerRequest(long elements) be renamed to SubscriberPuppet.ensureDemand(long elements), and that the contract be documented to say that when this method is invoked, the subscriber must eventually request at least elements number of elements, but it may request those elements in batches of one or more, and is allowed to wait for the requested elements to be received by the onNext method before requesting the next batch. Then the TCK tests should be modified to, instead of just asserting that there was a request, read the request, and then send the number of elements that were requested, looping if necessary to read more requests if the full amount of demand wasn't requested.

[viktorklang]

@jroper Why introduce a new method for this? Or am I missing something crucial?

[jroper]

It's not introducing a new method, it's renaming a method to match its intent. The method may not necessarily trigger a request for that many elements, but it must ensure that there is eventually at least enough demand for that many elements.

[viktorklang]

If we can rename it without breaking BC that's completely fine by me, as well as adding better docs for it.
I don't have enough spare cycles ATM so anyone, feel free to propose a PR.

[jroper]

It's a Java interface, you can't add an abstract method without renaming.

But also note, renaming is not the most important thing here, the important thing is to fix the broken tests. For example:

  @Override @Test
  public void required_spec308_requestMustRegisterGivenNumberElementsToBeProduced() throws Throwable {
    subscriberTest(new TestStageTestRun() {
      @Override
      public void run(WhiteboxTestStage stage) throws InterruptedException {
        stage.puppet().triggerRequest(2);
        stage.probe.expectNext(stage.signalNext());
        stage.probe.expectNext(stage.signalNext());

It needs to wait until a request for at least one element has been made before signalling the first mesage, (this particular issue is already reported in #277), but it then needs to wait until it has requests for at least 2 elements (the first request may only have asked for one, because some subscribers only support requesting one element at a time) before sending the second.

[viktorklang]

If we can change the method name without breaking backwards compat then I'm all for it. Otherwise a compromise would be to fix the semantics of the method and properly document that and that it has been changed.

Would you be able/willing to propose a PR to fix this?

[viktorklang]

@jroper Ping

[viktorklang]

@jroper Ping

[viktorklang]

Ping timeout