Whenever i use both subscribeOn and publishOn nothing is printed. If I use only one it will print. If I use subscribeOn(Schedulers.immediate()) or elastic it works. Any ideea why that is?
It was my understanding that publishOn affects on what thread it gets published and subscribe on on what thread the subscriber runs. Could you please point me in the right direction?
fun test() {
val testPublisher = EmitterProcessor.create<String>().connect()
testPublisher
.publishOn(Schedulers.elastic())
.map { it ->
println("map on ${Thread.currentThread().name}")
it
}
.subscribeOn(Schedulers.parallel())
.subscribe { println("subscribe on ${Thread.currentThread().name}") }
testPublisher.onNext("a")
testPublisher.onNext("b")
testPublisher.onNext("c")
Thread.sleep(5000)
println("---")
}
subscribeOn
rather influence where the subscription occurs. That is, the initial event that triggers the source to emit elements. The Subscriber
's onNext
hook on the other hand is influenced by the closest publishOn
up in the chain (much like your map
).
But EmitterProcessor
, like most Processor
s, is more advanced and can do some work stealing. I'm unsure why you don't get anything printed in your case (your sample converted to Java works on my machine), but I bet it has something to do with that Processor.
This code would better demonstrate subscribeOn
vs publishOn
:
Flux.just("a", "b", "c") //this is where subscription triggers data production
//this is influenced by subscribeOn
.doOnNext(v -> System.out.println("before publishOn: " + Thread.currentThread().getName()))
.publishOn(Schedulers.elastic())
//the rest is influenced by publishOn
.doOnNext(v -> System.out.println("after publishOn: " + Thread.currentThread().getName()))
.subscribeOn(Schedulers.parallel())
.subscribe(v -> System.out.println("received " + v + " on " + Thread.currentThread().getName()));
Thread.sleep(5000);
This prints out:
before publishOn: parallel-1
before publishOn: parallel-1
before publishOn: parallel-1
after publishOn: elastic-2
received a on elastic-2
after publishOn: elastic-2
received b on elastic-2
after publishOn: elastic-2
received c on elastic-2
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With