时间戳

首页 > 解决方案 > RxJava：如何刷新定时缓冲区？

rx-java - RxJava：如何刷新定时缓冲区？

问题描述

我正在创建这个 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 秒。我怎么做？

标签： rx-javarx-java2rx-android

解决方案

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(...)

推荐阅读