结合多个 Firebase 回调与 RxJava2

Question

首先，这是纯粹的地狱。我从来没有像回调地狱那样艰难。我恨它。总体思路是我有 Firebase 存储和 Firestore 云设置。 Storage 保存图像，而 Cloud 保存引用存储中图像位置的模型。然后我必须从存储中获取图像 url 以便我可以使用它并使用 glide 下载它。示例：Firebase 云为我提供了一个 drillType，其中包含 (title = "blabla", imageUrl = "scg/asd/asd")，然后我将 "scg/asd/asd" 的图像 Url 提供给我的远程存储并将其替换为我可以使用的 URL。问题是回调。我无法让它按照预期的方式发生。我试过暂停函数、观察者等等。这个想法是，onNext 应该调用 getImageUrl 并使用新的 Url 而不是存储位置更新 DrillType。但目前我无法让 onNext（或 .map）在继续之前等待执行 getImageUrl。

存储库目前不起作用

class DrillRepository(
    private val remoteDataSource: DataSource,
    private val remoteFilesDataSource: ImageUrlDataSource
) : CoroutineScope {


    private val job = Job()
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.IO + job


    fun loadDrillTypes(): Observable<DrillsType> {

        return Observable.create<DrillsType> { emitter ->
            val observer = object : DisposableObserver<DrillsType>() {
                override fun onError(e: Throwable) {
                    Log.d("TAG", "error " + e.message)
                    emitter.onError(e)
                }

                override fun onNext(data: DrillsType) {
                    data.let {
                    val singleObserver = object : SingleObserver<String>{
                        override fun onSuccess(t: String) {
                            it.drillType_imageUrl = t
                            dispose()
                        }

                        override fun onSubscribe(d: Disposable) {

                        }

                        override fun onError(e: Throwable) {
                            dispose()
                        }

                    }
                        getImageUrl(it.drillType_imageUrl).subscribeWith(singleObserver)
                    }
                }

                override fun onComplete() {
                    Log.d("TAG", "COMPLETE")
                    emitter.onComplete()
                    job.cancel()
                    dispose()
                }
            }

            remoteDataSource.loadDrillTypes().subscribeWith(observer)
        }
    }

fun getImageUrl(storageLocation: String): io.reactivex.Single<String> {
        return remoteFilesDataSource.getImageUrl(storageLocation)
    }

class FirebaseStorageDataSource(val firebaseStorage: FirebaseStorage) : ImageUrlDataSource {

    override fun getImageUrl(storageLocation: String): Single<String>
    {
        return Single.create<String>{emitter->
        firebaseStorage.getReferenceFromUrl(storageLocation).downloadUrl.addOnSuccessListener {
            emitter.onSuccess(it.toString())
        }.addOnFailureListener {
            emitter.onError(it)
        }
    }

class FirebaseFirestoreDataSource(val firebaseFirestore: FirebaseFirestore) : DataSource {


    override fun loadDrillTypes(): Observable<DrillsType> {

        return Observable.create<DrillsType> { emitter ->
            firebaseFirestore.collection("drilltypes")
                .get()
                .addOnSuccessListener { documents ->
                    for (document in documents) {
                        try {
                            val doc = document.toObject(DrillsType::class.java)
                            emitter.onNext(doc)
                        } catch (e: Exception) {
                            emitter.onError(e)
                        }
                    }
                    emitter.onComplete()
                }
                .addOnFailureListener { exception ->
                    Log.w("TAG", "Error getting documents: ", exception)
                    emitter.onError(exception)
                }
        }
    }

data class DrillsType(
    var drillType_title: String = "",
    var drillType_imageUrl: String = ""
)

我正在获取 DrillType，但我无法更新它们的图像Url。我不知何故需要等待执行，即来自 getImageUrl 的回调，然后使用更新的 drillType.

继续 onNext

编辑：这个问题我遇到过很多次了。这是迄今为止最糟糕的一次。我正在寻找一种可能而且最好是优雅的方式来解决这个问题。此外，观察者内部的观察者绝对是疯了，但那是我经过多次试验和错误后最终得出的结论。

编辑2： 解决方案，希望对未来的开发者有所帮助

    fun loadFullDrillType(): Observable<DrillsType> {
        return loadDrillTypeWithRawImageUrl().flatMapSingle { drillTypeWithRawImageUrl ->
            loadImageUrl(drillTypeWithRawImageUrl.drillType_imageUrl).map {
                return@map drillTypeWithRawImageUrl.copy(drillType_imageUrl = it)
            }
        }
    }

    fun loadDrillTypeWithRawImageUrl(): Observable<DrillsType> {
        return remoteDataSource.loadDrillTypes()
    }

        return Observable.create<DrillsType> { emitter ->
            firebaseFirestore.collection("drilltypes")
                .get()
                .addOnSuccessListener { documents ->
                    try {
                        for (document in documents) {
                            val doc = document.toObject(DrillsType::class.java)
                            emitter.onNext(doc)
                        }
                    } catch (e: Exception) {
                        emitter.onError(e)
                    }
                    emitter.onComplete()
                }
                .addOnFailureListener { exception ->
                    Log.w("TAG", "Error getting documents: ", exception)
                    emitter.onError(exception)
                }
        }
    }

Answer 1

要最大程度地使用 RX，每个 RX Firebase 包装器应该只执行一个操作（例如获取原始钻孔类型或只获取图像 URL）。然后你使用 RX 构造如 map 和 flatMap 来合并它们：

fun loadDrillTypeWithRawImageUrl(): Single<DrillsType> = TODO()
fun loadDrillImageUrl(rawImageUrl: String): Single<String> = TODO()

fun loadFullDrillType(): Single<DrillsType> {
    return loadDrillTypeWithRawImageUrl().flatMap { rawDrillsType ->
        loadDrillImageUrl(rawDrillsType.drillType_imageUrl).map { newImageUrl ->
            DrillsType(rawDrillsType.drillType_title, newImageUrl)
        }
    }
}

结合多个 Firebase 回调与 RxJava2

Combining multiple Firebase callbacks with RxJava2

android

kotlin

firebase

rx-java2

google-cloud-firestore