Combining multiple Firebase callbacks with RxJava2

Question

first off, this is pure hell. I was never having as hard time as with callback hell. I hate it. The general idea is that I have Firebase Storage and Firestore cloud setup. The Storage holds images, while the Cloud holds Models that have reference to image locations in the storage. Then I have to get the image url from the storage so I can use it and download it with glide. Example: Firebase cloud gives me a drillType with (title = "blabla", imageUrl = "scg/asd/asd"), I then give my remotestorage the imageUrl of "scg/asd/asd" and it replaces it with an URL that I can use. The problem are callbacks. I cannot get it to happen the way it's supposed to. I tried Suspend functions, Observers and so on. The idea is, onNext should call the getImageUrl and update the DrillType with the new Url instead of the storage location. But at the moment I cannot get the onNext (or .map) to wait for execution of the getImageUrl before going forward.

Repository CURRENTLY THIS DOESN'T WORK

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 = ""
)

I am getting the DrillTypes but I cannot update their ImageUrl. I somehow need to wait for the execution, ie the callbacks from the getImageUrl and then proceed with onNext using the updated drillType.

Edit: I've encountered this problem dozen of times. This has been the most awful one yet. I am looking for a possibly and preferably elegant way you guys solve this issue. Also the observer inside observer is absolutely crazy, but thats where I ended up after many trial and errors.

EDIT2: SOLUTION, hopefully, helpful to the future devs

    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

To use RX to the full extent, every RX Firebase wrapper should only perform one operation (such as fetching the original drill type or just fetching the image URL). Then you use RX constructs such as map and flatMap to merge them:

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)
        }
    }
}