RxJava:如何刷新定时缓冲区? [英] RxJava: how do you flush a timed buffer?
本文介绍了RxJava:如何刷新定时缓冲区?的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我正在创建这个 PublishProcessor,它每 10 秒将其元素保存到数据库中:
I'm creating this PublishProcessor that save to the database its element every 10 seconds:
val publishProcessor = PublishProcessor.create<Entity>()
publishProcessor
.buffer(10, SECONDS)
.observeOn(Schedulers.io())
.subscribe(
{ saveToDatabase(it) },
{ Log.e("TAG", "Error: $it") })
.addTo(compositeDisposable)
当我的活动完成时,我想刷新当前缓冲区中的所有内容,而不是等待 10 秒.我该怎么做?
When my activity finish, I want to flush everything that is in the current buffer, and not wait 10 seconds. How do I do that?
推荐答案
将另一个主题作为缓冲区边界并与一个区间合并:
Have another subject as the buffer boundary that is merged with an interval:
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);
如果您还想在手动冲洗时重置计时器
If you want to also reset the timer upon a manual flush
publishProcessor
.buffer(
flush.mergeWith(Observable.timer(10, TimeUnit.MILLISECONDS))
.take(1)
.repeat()
)
.observeOn(Schedulers.io())
.subscribe(...)
这篇关于RxJava:如何刷新定时缓冲区?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文