2.x Observable.observeOn() has an unbounded buffer

[adipascu]

Observable.observeOn() has an unbounded buffer and ignores the bufferSize parameter.

Why does Observable.observeOn() exist in the first place ? I think schedulers only make sense for Flowable.

This code crashes as expected:

 Flowable.create<Int>({
        for (i in 0..100) {
            it.onNext(i)
        }
    }, FlowableEmitter.BackpressureMode.NONE)
    .observeOn(Schedulers.computation(), false, 1)
    .subscribe {
        Thread.sleep(100)
    }

This does not crash, I think this should crash :

Observable.create<Int> {
        for (i in 0..1000) {
            it.onNext(i)
        }
    }
    .observeOn(Schedulers.computation(), false, 1)
    .subscribe {
        Thread.sleep(100)
    }

The code from above is written in Kotlin.

[akarnokd]

Observable.observeOn takes the bufferSize parameter to size its internal buffer's island size (SpscLinkedArrayQueue) to reduce allocation in case of a bursty source. Setting it to anything below 16 adds a relatively large overhead.

I think schedulers only make sense for Flowable

Crossing thread boundaries makes sense to all reactive base types so we have operators for it in every base reactive class.

[adipascu]

This is working as intended ? The documentation from Observable.observeOn() states:

Modifies an ObservableSource to perform its emissions and notifications on a specified Scheduler asynchronously with a bounded buffer of configurable size

From testing I never managed to hit this bounded buffer limit (the observable thread did not block) and I had a slow consumer.

[akarnokd]

That's a copy-paste error in the documentation. I'll fix it in my upcoming coverage PR.

[akarnokd]

Closing via #4730