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

Question

我正在开发连接到Firestore的Firebase Android应用程序。术语是该集合为“资产”。该示例代码具有简单的操作，例如addAsset和deleteAsset，它们可以正常工作：

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")。如果我在第二个记录命令处放置一个断点，它是开头的通常的红点，但是当调用进入该函数时，该图标将变为灰色的“请勿输入”图标，并且当我将鼠标悬停在该图标上时告诉我“在...找不到可执行文件”。因此，那里肯定有问题。

我是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的回答。

Answer 1

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

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

感谢Pavel指出这一点。