Using publish/share/etc leads to upstream operator running on unexpected scheduler #6144
Milestone
Comments
|
This a bug due to operator fusion breaking the thread-confinement expected after |
|
@akarnokd I've noticed that val scheduler1 = Schedulers.from(Executors.newSingleThreadExecutor { runnable ->
Thread(runnable, "scheduler1")
})
val scheduler2 = Schedulers.from(Executors.newSingleThreadExecutor { runnable ->
Thread(runnable, "scheduler2")
})
val flowable: Flowable<Long> = Flowable.interval(1L, TimeUnit.MILLISECONDS)
.onBackpressureLatest()
.take(10_000)
.observeOn(scheduler1)
.map {
val threadName = Thread.currentThread().name
if (threadName != "scheduler1") {
throw AssertionError("Wrong thread on map #1: $threadName")
}
it
}
.hide()
.share()
.map {
val threadName = Thread.currentThread().name
if (threadName != "scheduler1") {
throw AssertionError("Wrong thread on map #2: $threadName")
}
it
}
flowable
.observeOn(scheduler2)
.doOnNext { Thread.sleep(10L) }
.subscribe {
println("onNext on ${Thread.currentThread().name}: $it")
}Output: I have two questions about this:
|
No, this is a different effect. Due to backpressure and buffering, the
If you want that |
|
Thanks for the reply, @akarnokd. That makes sense - I will keep that in mind. Are there any existing mechanisms that allow something like |
|
No. It is an async boudary in which case all bets are off. You have to ensure the proper thread via observeOn. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
There appears to be a bug, or at the very least an undocumented and unexpected case, where
FlowablePublishcauses upstream operators to be applied on the subscription scheduler rather than the upstream observation scheduler (due topolling). This is reproducible at least in2.1.xand2.2.0.Sample (Kotlin)
Expected Behavior
It should be expected that the
.mapoperator is applied on thescheduler1scheduler since it is the scheduler supplied toobserveOnimmediately upstream.Actual Behavior
After about ~100 emissions, the
mapoperator is applied on thescheduler2scheduler. I'm not an expert in the details, but it seems to have something to do with backpressure, buffering, and polling for items upon subscription?Output
The text was updated successfully, but these errors were encountered: