Understanding flowable backpressure rxjava2

Question

I put together this dummy example to try to understand backpressure a little better:

Flowable.range(1, 100).onBackpressureDrop()
                      .subscribeOn(Schedulers.io())
                      .observeOn(AndroidSchedulers.mainThread())
                      .subscribeWith(object : DisposableSubscriber<Int>() {
                        override fun onStart() {
                          request(1)
                        }

                        override fun onComplete() {
                          Log.d(this@MainActivity::class.java.simpleName, "onComplete")
                        }

                        override fun onNext(t: Int?) {
                          Log.d(this@MainActivity::class.java.simpleName, t.toString())
                          Thread.sleep(1000)
                          request(1)
                        }

                        override fun onError(t: Throwable?) { //handle error}
                      })

I have an extremely slow Subscriber that consumes data from a very fast Flowable. And I'm instructing the Flowable to onBackPressureDrop(). Despite of this, my output looks like this (from 1 to 100)

07-16 23:07:21.097 22389-22389 D: 1
07-16 23:07:22.100 22389-22389 D: 2
07-16 23:07:23.102 22389-22389 D: 3
07-16 23:07:24.104 22389-22389 D: ...
07-16 23:07:24.104 22389-22389 D: ...
07-16 23:07:24.105 22389-22389 D: 99
07-16 23:07:25.105 22389-22389 D: 100
07-16 23:07:25.107 22389-22389 D: onComplete

I was expecting missing elements since the subscriber is extremely slow, but that isn't the case, all number from 1 to 100 are printed to the console, one every second.

Next, I tried to request all values at once. So I replaced request(1) in onStart with request(Long.MAX_VALUE) and remove request(1) from the onNext call. but it still prints the number 1 to 100 with no missing elements.

So I wonder how can I simulate a subscriber missing events for a slow subscriber? How can I make a backpressure exception happen?

Thanks

akarnokd akarnokd · Accepted Answer · 2017-07-17T10:25:28

observeOn has a default internal buffer size of 128, that's why you don't see elements dropped as it can simply buffer up all the 100 elements you are generating. You can set the buffer size to 1 via observeOn(mainThread(), false, 1) and experience drops.

Understanding flowable backpressure rxjava2

1 Answers