RxJava Flowable.create(), how to respect subscribeOn() thread

user_2051647 0
Viewed 1065

Im wrapping custom library (dataClient) callback api to RxJava Flowable. The dataClient uses its own threading, so it's callback is called on its own thread.

In my Rx chain, I try to specify computation scheduler using .subscribeOn(Schedulers.computation()). Still, when I print thread name on my Rx chain, im getting my dataClient thread.

What I should do, to make my Flowable to use thread specified in .subscribeOn()?

Flowable.create({ emitter ->
    dataClient.setCallback(object : Callback {
        override fun message(message: DataModel) {
            emitter.onNext(vehicle)
        }

        override fun done() {
            emitter.onComplete()
        }
    })
    emitter.setCancellable {
        dataClient.setCallback(null)
    }
}, BackpressureStrategy.BUFFER)
    .subscribeOn(Schedulers.computation())
    .doOnNext { Log.e("DATA", Thread.currentThread().name) }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe { data -> Log.d("DATA", "Got data" + data.id)) }
rx-java2 kotlin android java
1 Answers

The subscribeOn scheduler ensures that the subscription is done on the relating thread. Subscription happens exactly ones and it's handled differently than the observeOn scheduler, which schedules the emission of the element on the new thread.

Flowable.create({ emitter ->
    // this runs with the computation scheduler
    dataClient.setCallback(object : Callback {
        override fun message(message: DataModel) {
            // this runs on the thread it's called from
            emitter.onNext(vehicle)
        }

        override fun done() {
            // this runs on the thread it's called from
            emitter.onComplete()
        }
    })
    emitter.setCancellable {
        dataClient.setCallback(null)
    }
}, BackpressureStrategy.BUFFER)
    .subscribeOn(Schedulers.computation())
    .doOnNext {
        // this runs on the thread of the onNext call
        Log.e("DATA", Thread.currentThread().name)
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe {
        // this runs on the main thread
        data -> Log.d("DATA", "Got data" + data.id))
    }

Since your subscription code is not blocking and doesn't maintain the thread for emission, setting up subscribeOn is not necessary and could be omitted. It mostly only has effect with synchronous sources.

user_33852120
Related