How to buffer only latest emission from rx.Observable during backpressure

Question

I have an rx.Observable which emits the progress of a task to onNext(). The onNext() emissions can sometimes occur so quickly that the Observer cannot keep up, resulting in backpressure. I would like to handle the backpressure by only buffering the latest emission from the Observable.

For example:

• Observable emits 1 and Observer receives 1.
• While Observer is still processing 1, Observable emits 2, 3, and 4.
• Observer finishes processing 1 and begins processing 4 (emissions 2 and 3 are dropped).

This seems like it would be a common case for handling progress in an Rx Observable since you usually only care about updating your UI with the latest progress information. However I have not been able to figure out how to do this.

Anyone know how this can be achieved with RxJava?

Quadrivium

Answer 1

onBackPressureLatest is your friend here. :) http://reactivex.io/RxJava/javadoc/rx/Observable.html#onBackpressureLatest()

Samirasamisen

Thanks! That's exactly what I was looking for! Apparently I needed to update RxJava. :-P As an added note, I had to call this operator between subscribeOn(Schedulers.io()) and observeOn(AndroidSchedulers.mainThread()) to get it to work. – Quadrivium

Can we get an update on this for RxJava2? I'm trying to use toFlowable(BackpressureStrategy.LATEST) but it seems to be queueing up all requests and processing those as well. – Hatter

Answer 2

Observable.debounce sounds like what you need. In the example below the latest emission only from observable in each 200ms window will be sent to the observer.

observable
    .debounce(200, TimeUnit.MILLISECONDS)
    .subscribe(observer);

Immensity

This would work, think I think onBackPressureLatest is closer to the original question. debounce is bound to the given time frame, which is great for things like UI input. – Samirasamisen