通过订阅/取消订阅从侦听器发出项目的正确方法

在MVP架构之后，视图是一个Android活动，每个人都有自己的主持人。

作为数据存储库，我有一个图层，每次在后端发生变化时都会通知我新的数据。 存储库可以由不同的演示者使用，演示者必须在视图被销毁时销毁。

如何实现一个响应式流（每个存储库一个），演示者可以订阅和取消订阅以获取数据更新？

其他要求是流应该能够结合并应用它们的反应操作符。

我通过Firebase Firestore后端尝试了以下内容，但发现内存泄漏（活动和演示者不会被破坏）。

注意：当后端发生变化时，addSnapshotListener（）回调会通知新数据。

正如您所看到的，observable是公开的（默认情况下是Kotlin），供演示者使用，用于订阅和取消订阅。

class DocumentRepository( path: List, private val model: Class) { private var documentReference: DocumentReference val observable: Observable private var emitter: ObservableEmitter? = null private lateinit var item: T init { documentReference = FirebaseFirestore.getInstance().collection(path[0]).document(path[1]) for (i in 2..path.lastIndex step 2) documentReference = documentReference.collection(path[i]).document(path[i + 1]) observable = Observable.create(this::listenChanges) } private fun listenChanges(emitter: ObservableEmitter) { this.emitter = emitter documentReference.addSnapshotListener { documentSnapshot, _ -> item = documentSnapshot.toObject(this.model) this.emitter?.onNext(item) } } fun get() { emitter?.onNext(item) } fun put(item: T) { item.updatedAt = TimeExtension.now() documentReference.set(item) } fun delete(item: T) { documentReference.delete() } }