Reactor-core: README.md describes EmitterProcessor behavior incorrectly

[a-voloshyn]

See https://github.com/reactor/reactor-core/blame/5bdac9a12e7bc6a294a516d490470008b66bcb4d/README.md#L153-L161

EmitterProcessor<Integer> emitter = EmitterProcessor.create();
BlockingSink<Integer> sink = emitter.connectSink();
sink.next(1);
sink.next(2);
emitter.subscribe(System.out::println);
sink.next(3); //output : 3
sink.finish();

I tried to execute the code snippet using io.projectreactor:reactor-core:3.0.5.RELEASE and it failed with an exception:

Exception in thread "main" reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)
    at reactor.core.Exceptions.failWithOverflow(Exceptions.java:140)
    at reactor.core.publisher.Operators.reportMoreProduced(Operators.java:477)
    at reactor.core.publisher.BlockingSink.next(BlockingSink.java:160)

It works as described, if I add a dummy (skipping) subscriber before the sink invocations:

EmitterProcessor<Integer> emitter = EmitterProcessor.create();

// Subscribe before the emitting
emitter.subscribe(i -> {});

BlockingSink<Integer> sink = emitter.connectSink();
sink.next(1);
sink.next(2);
emitter.subscribe(System.out::println);
sink.next(3); //output : 3
sink.finish();

[Kindrat]

+1

[apischan]

As I see that this issue labeled as documentation but IMHO this is the issue in code rather than in documentation.
I have checked similar example in RxJava and it works without exception.

PublishSubject<Integer> behaviorSubject = PublishSubject.create();
behaviorSubject.onNext(1);
behaviorSubject.onNext(2);
behaviorSubject.subscribe(System.out::println);
behaviorSubject.onNext(3);
behaviorSubject.onCompleted();

[smaldini]

Indeed, just noting that PublishSubject is unbounded so better fit in rx is PublishProcessor. However we are reworking and improving this part as #248 and will mark that BlockingSink dance unnecessary . Thats the one failing with overflow here since there is effectively no request yet to the emitter as nothing is subscribed. Technically Emitter could just drop in that case.

[smaldini]

EmitterProcessor does not require connect anymore and offers a sink instead of connectSink to mirror Flux.create FluxSink API instead of inventing a new one BlockingSink.
I'll move the issue in M2 since the recent API polishes pre Spring 5 RC1 requires us to release asap the binary and we can resume on introducing/documenting processor in 3.1.0.M2

[simonbasle]

Closing this as it has been reworked in #525. @a-voloshyn the reference documentation will give details on Processors usage by 3.1.0.RELEASE (still pending). @apischan if the issue you encountered is not fixed in 3.1.0.M1, please reopen this issue or open a new one, at your convenience.