RxJava: как вы очищаете синхронизированный буфер?

Question

Я создаю этот PublishProcessor, который каждые 10 секунд сохраняет в базе данных свой элемент:

    val publishProcessor = PublishProcessor.create<Entity>()

    publishProcessor
        .buffer(10, SECONDS)
        .observeOn(Schedulers.io())
        .subscribe(
            { saveToDatabase(it) },
            { Log.e("TAG", "Error: $it") })
        .addTo(compositeDisposable)

Когда моя деятельность заканчивается, я хочу сбросить все, что находится в текущем буфере, а не ждать 10секунд.Как мне это сделать?

Answer 1

В качестве границы буфера укажите другой объект, объединенный с интервалом:

PublishSubject<Entity> publishProcessor = PublishSubject.create();

Subject<Long> flush = PublishSubject.<Long>create().toSerialized();

publishProcessor
    .buffer(flush.mergeWith(Observable.interval(10, TimeUnit.MILLISECONDS)))
    .observeOn(Schedulers.io())
    .subscribe(...)

flush.onNext(1L);

Если вы также хотите сбросить таймер при ручной очистке

publishProcessor
    .buffer(
        flush.mergeWith(Observable.timer(10, TimeUnit.MILLISECONDS))
        .take(1)
        .repeat()
    )
    .observeOn(Schedulers.io())
    .subscribe(...)