Reactor-core: CancellationException occurs while subscribing to connectable hot source

[stef2georg]

I think the following scenario introduces a bug.

Expected behavior

test
test

Actual behavior

test
Exception in thread "main" reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.concurrent.CancellationException: Disconnected
Caused by: java.util.concurrent.CancellationException: Disconnected
at reactor.core.publisher.FluxReplay$ReplaySubscriber.dispose(FluxReplay.java:1202)
at reactor.core.publisher.OperatorDisposables.dispose(OperatorDisposables.java:132)
at reactor.core.publisher.FluxRefCount$RefCountMonitor.innerCancelled(FluxRefCount.java:132)
at reactor.core.publisher.FluxRefCount$RefCountInner.cancel(FluxRefCount.java:200)
at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:75)
at reactor.core.publisher.FluxRefCount$RefCountInner.onNext(FluxRefCount.java:177)
at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:808)
at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay(FluxReplay.java:892)
at reactor.core.publisher.FluxReplay.subscribe(FluxReplay.java:1085)
at reactor.core.publisher.FluxRefCount$RefCountMonitor.subscribe(FluxRefCount.java:116)
at reactor.core.publisher.FluxRefCount.subscribe(FluxRefCount.java:77)
at reactor.core.publisher.MonoNext.subscribe(MonoNext.java:40)
at reactor.core.publisher.Mono.subscribe(Mono.java:3077)
at reactor.core.publisher.Mono.subscribeWith(Mono.java:3185)
at reactor.core.publisher.Mono.subscribe(Mono.java:3071)
at reactor.core.publisher.Mono.subscribe(Mono.java:3038)
at reactor.core.publisher.Mono.subscribe(Mono.java:2985)
at test.Test.main(Test.java:10)

Steps to reproduce

Flux<String> flux = Flux.<String>create(sink -> sink.next("test"))
    .replay(1)
    .refCount();

flux.subscribe(System.out::println);
flux.next().subscribe(System.out::println); // The exception is thrown here!

Reactor Core version

3.1.4

JVM version (e.g. java -version)

java version "1.8.0_121"
Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)

I also posted a question to Stackoverflow.

[rotilho]

refCount disconnect after all subscriber cancel or upstream is completed.

You need to use autoConnect instead in this case.

[stef2georg]

Hi,

Thanks for your answer. I agree that refCount disconnects, but the flux will be recreated once there is a new subscriber. A quote from the documentation:

refCount(n) not only automatically tracks incoming subscriptions but also detects when these subscriptions are cancelled. If not enough subscribers are tracked, the source is "disconnected", causing a new subscription to the source later if additional subscribers appear.

Consider the following example (it works as expected):

Flux<String> flux = Flux.<String>create(sink -> sink.next("test"))
            .replay(1)
            .refCount();

    flux.subscribe(System.out::println);
    flux.subscribe(System.out::println);
    flux.next().subscribe(System.out::println);

Output:

test
test
test

The described problem happens only when we have single Flux.subscribe() followed by a Mono.subscribe().

[ivailo-pashov]

I agree with @stef2georg it seems that refCount() operator implementation does not handle properly early unsubscribe while it is emitting its next event.

[simonbasle]

I need to investigate this further, as there might be an issue with refCount, but in the meantime I'd like to point out two issues with your example:

• your create never calls onComplete
• you don't implement an error callback in your subscribes.

You don't see the second test because of that second point: not having a callback will default to throwing a CallbackNotImplementedException, not giving enough time to the Mono subscriber to print the event it _did_ receive. Adding an error handler will print both the disconnect error (from the Flux subscribe) AND the mono's test.

If you fix the create to call complete(), both your snippets work as expected.

Investigating further if there is a subscriber-tracking issue in FluxRefCount.