观察者订阅我的 Observable 时如何返回默认值

Question

我还是 RxJava 的新手，并试图用它来实现一些东西。

我的 Android 应用程序中有通知计数器，我想每五分钟从 API 获取计数，我想将此计数共享给多个活动。所以我想将计数存储在管道中，以便每个监听该管道的活动都可以获得最后发出的值。

我从一个 Object 类开始，添加了两个 BehaviorSubjects 一个用于提供默认值，第二个用于在 5 分钟完成时发出真值“告诉正在侦听的活动是否有”您应该获取新计数.

但问题是，当我收听 BehaviorSubject 时，它会发出管道中的所有数据！我怎样才能只得到最后一个

object RxBus {

@JvmStatic
var count: Int = 0

@JvmStatic
private var fetchData: Boolean = true

@JvmStatic
private var behaviorSubject: BehaviorSubject<Any> = BehaviorSubject.createDefault(CountEvent(count))

@JvmStatic
private var fetchSubject: BehaviorSubject<Any> = BehaviorSubject.createDefault(FetchEvent(fetchData))

private val FIVE_MINUTES: Long = 1000 * 60 * 5

init {
    fixedRateTimer("countTimer", false, 0L, FIVE_MINUTES) {
        fetchSubject.onNext(FetchEvent(true))
    }
}

fun publish(event: Any) {
    if (event is CountEvent) {
        Log.d(TAG, "eventCount = ${event.count}")
        count = event.count
        fetchData = false
    }
}

fun listenToLoadEvent(): Observable<FetchEvent> {
    return fetchSubject.ofType(FetchEvent::class.java)
}


fun listen(): Observable<CountEvent> {
    return behaviorSubject.ofType(CountEvent::class.java)
}

请注意，我想在每个活动的onResume 中收听两个主题。有没有更好的方法来做到这一点？如果不能，你能帮我只得到最后一个发出的项目。

提前致谢

Answer 1

    class CountEvent(val value: Int)

    object CounterManager {
        val counterDataSource = BehaviorProcessor.create<CountEvent>()

        val refreshNotifier = interval(5, TimeUnit.MINUTES).publish()
                .refCount()

        fun publish(event: Any) {
            if (event is CountEvent) {
                counterDataSource.onNext(event)
            }
        }

    }