可以观察到，当没有订阅服务器时缓冲项，然后发出它们，并在订阅时清除缓冲区？

Question

我想要一个可以观察到的：

1. 可以按需发出项目，但从未真正完成（一个热的可观察的？）
2. 知道何时有订阅者
3. 如果没有订阅服务器，它将缓冲我告诉它发出的项
4. 订阅后，它将按顺序发出缓冲项，然后清除缓冲区，然后继续允许我发出更多项
5. 取消订阅时（订阅服务器已释放？），它将返回缓冲。

这是我所想的一个伪代码--我没有必要的回调来正确地执行此操作。另外，如果我能把它全部用一个可观察的或主题来包装，那就好了。

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。如果您担心缓冲区变得太大，它有无界版本和有界版本。