How to fetch the API response while using Observable.zip(...)?

Question

I have to hit 3 API's to update the same screen so for this i think RxJava is the fastest way to do that in parallel. While i was searching for the implementation i came across Observable.zip(...) function as it can perform multiple API hits in parallel.

I am using Retrofit for calling API's and have already created Pojo class with gson annotation.

Sample Pojo classes:

data class ResponseGetFCData(

    @SerializedName("End")
    val end: String,
    @SerializedName("Uni")
    val uni: String,
    @SerializedName("Y")
    val y: Double
)

data class ResponseAK(

    @SerializedName("End")
    val end: String,
    @SerializedName("Manu")
    val manu: String,
    @SerializedName("Start")
    val start: String,
    @SerializedName("TY")
    val tY: Double

)

Sample Api Interface:

interface Api{
@GET("GetUniPI")
    fun getFCdata(@Query("pi") pi: String
                 , @Query("uni") uni: String): Observable<ResponseGetFCData>
}

Objective : From the response of 2 out of 3 API's I have to compute some mathematical calculation and the third API response will carry data for recycler view. Here i have to compute (y * ty)/100 by taking y from API 1 and ty from API 2 and such similar computations.

MyCode: In activity onCreate(....):

        val requests = ArrayList<Observable<*>>()
        val backendApi = WinRetrofitHelper.winApiInstance()

        requests.add(backendApi.getFCdata("","","",""))
        requests.add(backendApi.getAKCountry())
        requests.add(backendApi.getRecyclerData("","",""))

        Observable
                .zip(requests) {

                }
                )
                .subscribe({

                    Log.e("Exe Summary","******************Success*******************")

                }) {

                    Log.e("Exe Summary",it.stackTrace.toString())
                }

So here i am not getting how to fetch the response from these 3 API's and how and where to compute the maths and how will i update the data in recyclerview adapter from 3rd API response. Please help me to understand this with a better approach.

Answer 1

Try using below:

Observable.zip(
    backendApi.getFCdata("","","",""),
    backendApi.getAKCountry(),
    backendApi.getRecyclerData("","",""),
    Function3<ResponseGetFCData, ResponseAK, List<ResponseMarket>, List<ResponseMarket>> { 
        fcData, akCountry, recyclerData ->

        // Your operation here

        return recyclerData
    })
    .observeOn(AndroidSchedulers.mainThread())
    .doOnSubscribe { /* Loading Start */ }
    .doOnTerminate { /* Loading End */ }
    .subscribe(
            { /* Successfully Synced */ },
            { /* Having error */ }
    )

Answer 2

Or you can give coroutines a try. It has simple syntax easy to understand

fun toDoWorkConcurrent() {

    job2 = launch {

        try {
            val work1 = async { getThingsDone(43) }
            val work2 = async { getThingsDoneAgain(123) }

            val result = computeResult(work1.await(), work2.await())

            withContext(UI) {
                tvResult1.text = result.toString()
            }

        } catch (exception: Exception) {
            exception.printStackTrace()
        }

    }
}

private fun computeResult(await: Int, await1: Int): Int {
    return await + await1
}

Edit: Source

Answer 3

Please try like this

 Observable.zip(yourobservalelist, new Function<Object[], Object>() {
        @Override
        public Object apply(Object[] objects) throws Exception {


            return objects;
        }
    })

            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(Disposable disposable) throws Exception {

                }
            })
            .doOnTerminate(new Action() {
                @Override
                public void run() throws Exception {

                }
            })
            .subscribe(new Consumer<Object>() {
                           @Override
                           public void accept(Object o) throws Exception {
                               //Do something on successful completion of allrequests

                               //}
                           }
                       },

                    // Will be triggered if any error during requests will happen
                    new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable e) throws Exception {
                            //Do something on error completion of requests
                            e.printStackTrace();
                        }
                    });


}