RxJava2 onBackpressureBuffer() действие, параметры, буфер?

Question

[foonfyrick]

В общем, есть такой код:

MyRetrofit.api.getPostsFlowable()
            .onBackpressureBuffer()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
            Log.e("onFlowable",it.toString())
        }

Я получаю список данных с https://jsonplaceholder.typicode.com/posts
Весь этот список я вывожу в логи.
В этом примере меня интересует строка onBackpressureBuffer(), потому что как я ни старался, не вижу как она работает.
Я прочитал, то что буфер это - линейная последовательность данных примитивного типа, как я понял, это те данные, которые я вывожу. В observeOn поумолчанию буфер = 128(чего-то там, так и не понял чего), так вот, в onBackpressureBuffer() можно задать параметры:
1) capacity() - размер буфера при достижении которого будет уведомление о BufferOverflowException, но, какое-бы я число не ставил, ничего нигде не появляется.
2) BackpressureOverflowStrategy - стратегия, определяющая поведение при достижении заданной емкости буфера.
Так вот, как мне отредактировать выше-написанный код, чтобы я увидел, что происходит, если буфер доходит до указанной отметки? Как его переполнить, чтобы я увидел разницу с .onBackpressureBuffer() и без?
Спасибо.

Answer 1

[Денис Загаевский]

Backpressure это такая ситуция, когда сабскрайбер(консьюмер) потребляет(обрабатывает) значения медленнее, чем эмиттер(продьюсер) их производит. Тогда события, конечно, никуда деться не могут, их надо копиь в буфер, сбрасывать(drop), или сразу при возникновении backpressure кидать ошибку. В твоей ситуации явно никакого backpressure случиться не может, тк ты просто берешь данные из сети(долго) и выводишь их в лог(быстро).
Смоделировать долгую обработку можно, сделав искусственную задержку(Thread.sleep) и сделав быстрый эмит. Например, для наглядности, создай фловабл, и закинь в него 10000 элементов в цикле. А в сабскрайбе - слип.

Comments

[foonfyrick]

В общем, я всяко пытался написать, и с sleep и без, единственный вариант, когда вывод отличается от ожидаемого, это

val flowable = Flowable.range(0,10000)
       .onBackpressureBuffer(1,{},BackpressureOverflowStrategy.DROP_OLDEST)
       .observeOn(Schedulers.computation()).subscribe{
           Log.e("@",it.toString())
       }

Здесь, при выводе, значения периодически пропадают, в других примерах, когда я ставил sleep, просто значения выходили медленно

flovable
            .onBackpressureBuffer(10)
            .subscribe{
            Thread.sleep(10000L)
            Log.e("@@@@@",it.toString())
        }

val flovable = Flowable.create(FlowableOnSubscribe<Int> {
        for(a in 1..10000){
        it.onNext(a)
        }
    },BackpressureStrategy.ERROR)

Странно, но нигде вообще нету примеров, чтобы было видно что происходит без контролирования размера буфера и с... , гуглю уже часа 3 часа...
Куда ошибка должна приходить, если strategy_ERROR ставить, тоже не понятно.
Написано -ERROR — это дефолтное поведение при всех предыдущих перегрузках, выбрасывающее BufferOverflowException.
Куда оно должно выбрасываться, в логах ничего нет, эмулятор работает.

[Денис Загаевский]

foonfyrick, ошибка выбрасывается в сам стрим

[foonfyrick]

Денис Загаевский, и размер буфера 128 это чего 128? 128 потребляемых значений?

[Денис Загаевский]

foonfyrick, да, это сколько он сохранит перед тем, как кинуть ошибку