kotlin - 可观察到在没有订阅者时缓冲项目，然后在订阅时发出它们并清除缓冲区？

Question

我想要一个可观察到的:

1. 可以按需发出项目并且永远不会真正完成(热可观察？)
2. 知道它何时有订阅者
3. 如果没有订阅者，它将缓冲我告诉它发出的项目
4. 当订阅时，它会按顺序发出缓冲的项目，然后清除缓冲区，然后继续让我发出更多的项目
5. 当取消订阅时(订阅者被处置？)，它会回到缓冲状态。

还有:

1. 一次只能有一个订阅者

2. 不需要线程安全

这是我所想的伪代码——尽管我没有必要的回调来以正确的方式执行此操作。另外，如果我可以将其全部包装在一个 Observable 或 Subject 中，那就太好了。

class RxEventSender {
    private val publishSubject = PublishSubject.create<Action>()

    val observable: Observable<Action> = publishSubject

    private val bufferedActions = arrayListOf<Action>()

    private var hasSubscribers = false

    fun send(action: Action) {
        if (hasSubscribers) {
            publishSubject.onNext(action)
        } else {
            bufferedActions.add(action)
        }
    }

    //Subject was subscribed to -- not a real callback
    fun onSubscribed() {
        hasSubscribers = true
        bufferedActions.forEach {action ->
            publishSubject.onNext(action)
        }
        bufferedActions.clear()
    }

    //Subject was unsubscribed -- not a real callback
    fun onUnsubscribed() {
        hasSubscribers = false
    }
}

Answer 1

使用 ReplaySubject .如果您担心缓冲区变得太大，它有无界和有界版本。