PublisherVerification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete fails if the publisher completes synchronously

[jroper]

Consider the following empty publisher:

class EmptyPublisher implements Publisher<Void> {
  public void subscribe(Subscriber<? super Void> subscriber) {
    Objects.requireNonNull(subscriber);
    subscriber.onSubscribe(new Subscription() { ... });
    subscriber.onComplete();
  }
}

I believe this is a valid publisher according to the spec, there's nothing that says that onComplete can't be synchronously invoked from subscribe, it just has to be invoked after onSusbscribe (it is).

However PublisherVerification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete, a test for empty publishers which should pass on the above publisher, fails with java.lang.AssertionError: Subscriber::onComplete() called before Subscriber::onSubscribe.

The problem seems to be that ManualSubscriberWithSubscriptionSupport requires expectCompletion to be invoked after onSubscribe and before onComplete is invoked. But expectCompletion isn't invoked until after subscribe returns, hence synchronous invocation of onComplete after onSusbcribe in the subscribe method will always fail.

[akarnokd]

(Oops, wrong 105 method).

RxJava and Reactive4JavaFlow both pass so I don't think there is anything nothing wrong with the TCK in this regard. Did you override maxElementsFromPublisher to return zero?

[akarnokd]

This passes (but skips 38 of the 42 tests due to max length == 0):

import org.reactivestreams.*;
import org.reactivestreams.tck.*;
import org.testng.annotations.Test;

@Test
public class EmptyPublisherTckTest extends PublisherVerification<Integer> {
    
    public EmptyPublisherTckTest() {
        super(new TestEnvironment(50));
    }

    @Override
    public Publisher<Integer> createPublisher(long elements) {
        return subscriber -> {
            subscriber.onSubscribe(new Subscription() {

                @Override
                public void request(long n) {
                }

                @Override
                public void cancel() {
                }
            });
            subscriber.onComplete();
        };
    }

    @Override
    public Publisher<Integer> createFailedPublisher() {
        return null;
    }

    @Override
    public long maxElementsFromPublisher() {
        return 0;
    }
}

[ktoso]

It's as @akarnokd writes. If your publisher is not able to emit many elements, you have to specify this by overriding the maxElementsFromPublisher method, so then tests which require elements to flow are skipped.

[viktorklang]

The error message could be better, perhaps add a hint if onComplete happens before expected? (please override / verify max number of elements)

…

-- Cheers, √

On Dec 29, 2017 12:46, "ktoso" ***@***.***> wrote: It's as @akarnokd <https://github.com/akarnokd> writes. If your publisher is not able to emit many elements, you have to specify this by overriding the maxElementsFromPublisher method, so then tests which require elements to flow are skipped. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#422 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAAqd58AIFpraGLrGxJnrSPEMYqiY_6bks5tFNEAgaJpZM4ROz7C> .

[jroper]

But optional_spec105_emptyStreamMustTerminateBySignallingOnComplete is not a test that requires elements to flow, it invokes createPublisher(0) to create an empty publisher, so if returning zero from maxElementsFromPublisher causes it to not run, then that's a bug too. The publisher I'm testing of course is more complex than this, I've just provided a minimal reproducer to simplify things, my publisher is definitely able to emit many elements, so it makes no sense to override maxElementsFromPublisher to return zero. It's just that, in the case that zero elements are requested, my publisher happens to know this already before subscribe is called that it is an empty publisher and so invokes onComplete synchronously from the subscribe method. And that's valid according to the spec, and so should pass the test.

[akarnokd]

The tests expect you to provide Publishers of certain sizes and won't pass if you provide something shorter. Also if your Publisher can have different behavior for different maxElementsFromPublisher, have multiple test classes for each case.

[jroper]

I understand that. Here's the test:

https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java#L412

Here's the code:

  @Override @Test
  public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable {
    optionalActivePublisherTest(0, true, new PublisherTestRun<T>() {
      @Override
      public void run(Publisher<T> pub) throws Throwable {
        ManualSubscriber<T> sub = env.newManualSubscriber(pub);
        sub.request(1);
        sub.expectCompletion();
        sub.expectNone();
      }
    });
  }

It's requesting a publisher of zero elements. This is a publisher of zero elements:

 Publisher publisher = subscriber -> {
            subscriber.onSubscribe(new Subscription() {
                public void request(long n) {}
                public void cancel() {}
            });
            subscriber.onComplete();
        };

It should pass that one test, right? Of course it won't pass the other tests, but it should pass the test that tests for an empty publisher, right? It's not passing.

[jroper]

I've created a PR that fixes the issue.

As you can see, in the PR, I've implemented a test that reproduces it - and in that test, as suggested, I did create a publisher that returned zero from maxElementsFromPublisher, and it still failed without my fix (you may not have noticed the failure when you ran it because the test is optional, and so you would have seen it as skipped - but skipped is a failure if something is supposed to pass it).