为什么 RxJava Completable 发射器中的 Firebase 任务不执行？答案

【问题标题】：Why Firebase Task in RxJava Completable emitter doesn't execute?为什么 RxJava Completable 发射器中的 Firebase 任务不执行？
【发布时间】：2020-08-02 11:03:56
【问题描述】：

我正在开发一个连接到 Firestore 的 Firebase Android 应用程序。命名法是集合是“资产”。示例代码具有简单的操作，例如 addAsset 和 deleteAsset，它们工作正常。这是实际与 Firebase 对话的数据存储层，视图模型层在此之上。

class FirestoreAssetRepository(secondaryDB: FirebaseFirestore) : IAssetRepository {

    companion object {
        private const val TAG = "FirestoreAssetRepo"
        private const val ASSET_COLLECTION = "Assets"
    }

    private var remoteDB: FirebaseFirestore
    private var changeObservable: Observable<List<DocumentSnapshot>>

    init {
        remoteDB = secondaryDB
    }

    override fun addAsset(asset: Asset): Completable {
        return Completable.create { emitter ->
            remoteDB.collection(ASSET_COLLECTION)
                .add(mapToAssetData(asset))
                .addOnSuccessListener {
                    if (!emitter.isDisposed) {
                        emitter.onComplete()
                    }
                }
                .addOnFailureListener {
                    if (!emitter.isDisposed) {
                        emitter.onError(it)
                    }
                }
        }
    }

    override fun deleteAsset(assetId: String): Completable {
        return Completable.create { emitter ->
            remoteDB.collection(ASSET_COLLECTION)
                .document(assetId)
                .delete()
                .addOnSuccessListener {
                    if (!emitter.isDisposed) {
                        emitter.onComplete()
                    }
                }
                .addOnFailureListener {
                    if (!emitter.isDisposed) {
                        emitter.onError(it)
                    }
                }
        }
    }

我正在向存储库添加一个可以修改特定文档的操作。

    override fun lockUnlockAsset(assetId: String): Completable {
        Log.d(TAG, "lockUnlockAsset")
        return Completable.create { emitter ->
            remoteDB.collection(ASSET_COLLECTION)
                .document(assetId)
                .get()
                .addOnSuccessListener {
                    Log.d(TAG, "Unlocking")
                    val remoteAsset = mapDocumentToRemoteAsset(it)
                    it.reference.update(getUnlockLocation())
                    if (!emitter.isDisposed) {
                        emitter.onComplete()
                    }
                }
                .addOnFailureListener {
                    Log.d(TAG, "Could not find asset to unlock")
                    if (!emitter.isDisposed) {
                        emitter.onError(it)
                    }
                }
        }
    }

执行到达Log.d(TAG, "lockUnlockAsset")，但从未到达Log.d(TAG, "Unlocking")。如果我在第二个日志记录命令处放置一个断点，它是开始时通常的红点，但是当调用进入函数时，图标变为灰色的“不要输入”图标，当我将鼠标悬停在它上面时 Android Studio告诉我“在...处找不到可执行文件”。所以肯定有问题。

我是 Kotlin 和 RxJava2 的新手。我怎样才能让它工作？

更新：回答 Pavel 的问题：这些函数是从 ViewModel 层调用的：

fun deleteAsset(assetId: String) {
    repository.deleteAsset(assetId)
        .subscribeOn(Schedulers.io())
        .subscribe(
            {},
            {
                it.printStackTrace()
            })
        .addTo(disposable)
}

fun addAsset(assetTitle: String) {
    repository.addAsset(Asset("${System.currentTimeMillis()}", assetTitle))
        .subscribeOn(Schedulers.io())
        .subscribe(
            {},
            {
                it.printStackTrace()
            })
        .addTo(disposable)
}

fun lockUnlockAsset(assetId: String) {
    repository.lockUnlockAsset(assetId)
}

我在存储库级别尝试使用.subscribeOn(Schedulers.io()).observe 的组合。也许是 .addTo(disposable) 让它工作了，我不确定我错过了什么。现在它正在工作，我等待 Pavel 的回答。

【问题讨论】：

你的失败监听器被调用了吗？如果是，那里的错误是什么？如果没有，也许你没有订阅Completable？
@xizzhu addOnFailureListener 有一个调试日志语句，我也没有在日志中看到它。奇怪的是为什么 Android Studio 报告我的内部指令是“找不到可执行文件”。我想我可能会错过一些东西，比如我必须有一个额外的观察者或听众链接到发声器上才能让它工作。或者其他一些系统性问题。但是addAsset 和deleteAsset 有效。
你能展示你如何执行lockUnlockAsset() Completable，我猜你还没有订阅
@PavelPoley 我正在尝试使用observeOn 和后续链调用，但我遗漏了一些东西，你的问题鼓励我更多地关注那个领域。我将用这些代码部分完成我的问题。所以现在它正在工作：我在调用存储库函数的 ViewModel 级别上缺少.subscribeOn(Schedulers.io()).subscribe({}, { it.printStackTrace() }).addTo(disposable) 的链调用。

标签： firebase kotlin google-cloud-firestore rx-java2

【解决方案1】：

我在数据存储库级别尝试了.subscribeOn(...) 和observeOn(..) + .observe(...) 的组合，但我应该只是遵循视图模型中的模式（视图模型调用数据存储库的函数）：它是链式subscribeOn + subscribe + addTo(disposable)：

fun lockUnlockAsset(assetId: String) {
    repository.lockUnlockAsset(assetId)
        .subscribeOn(Schedulers.io())
        .subscribe(
            {},
            {
                it.printStackTrace()
            })
        .addTo(disposable)
}

感谢 Pavel 指出这一点。

【讨论】：