CODEWITHSUNDEEP
androidkotlinrx-javareactive-programmingrx-kotlin
Jhon Fredy Trujillo Ortega
Jhon Fredy Trujillo Ortega

Reputation: 488

How to Merge multiple Observables with diferents call backs into one single Stream?

I have this problem. I'm trying to sync a local database into my android app with a remote app. I'm creating the logic to upload the new info created locally and the server responds with the remote id save it in the server. To archive this, I'm using a method that takes an array of objects an return an Observable which emit the response of the server for each element. Like this.

val dailyEntries = App.db.dailyEntryDao().getDailyEntriesCreated()
            dailyEntries.sync(context) //Return an observable
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe({                        
                        val response = DailyEntry(it)//Creates a daily entry using 
                                                      the response from the server
                        thread {                                
                          App.db.dailyEntryDao().update(response)
                        }
                    }, {
                        it.printStackTrace()
                    }, {
                        uploadEnclosures()
                    })

How you can see, in the onSuccess from the current observable calls another method. It uses the same logic and is showing ahead. 

private fun uploadEnclosures() {
        thread {
            val enclosures = App.db.enclosureDao().getEnclosuresCreated()
            enclosures.sync(context)
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe({
                        val response = Enclosure(it)
                        thread {                                
                            App.db.enclosureDao().update(response)
                        }
                    }, {
                        it.printStackTrace()
                    }, {
                        uploadSmokeTest()
                    })
        }
    }

It goes on with all the tables. Always we perform the update of the next table in the onSuccess of the current Observable. It is done like this because i need to make the sync in a specific order.

My question is, Is there a way to merge all these Observables in just one to perform a single Subscribe and controlling each onNext emotion?

Thanks for the answers

Upvotes: 0

Views: 1873

Answers (1)

rahul.taicho
rahul.taicho

Reputation: 1379

Well yes, but there would be a small bit of work required, you can use concat operator which takes care of the ordering for you and pass it a list of observables in order and then subscribe to it using a single observer which expects Any event to trickle down to it.

To be strict on type safety you can mark your observable source types with a common interface and use instance check to take actions specific to the event type.

Check more here

Code example -

fun concatCalls(): Observable<Any> {
    return Observable.concat(src1, src2, ...)
}

The consumer would look like this then -

concatCalls().subscribe(object: Subscriber<Any> {
    override fun onNext(o: Any) {
       when (o) {
           is Object1 -> // do handling for stuff emitted by src1
           is Object2 -> // do handling for stuff emitted by src2
           ....
           else // skip
    }
    ....
})

Upvotes: 2

Related Questions