Chaining requests in Retrofit + RxJava

Question

I have 2 APIs that I want to make request to in sequence and store their data in SQLite.

First I want to make request to API A and store its data in SQL table a. Then make request to API B and store its data in table b and some data in table a_b. The data stored in a_b is from request B alone.

How can I do this using RxJava. I read somewhere about using flatMap for this, something like this

apiService.A()
    // store in DB here? How? maybe use map()?
    .flatMap(modelA -> {
        // or maybe store modelA in DB here?
        return apiService.B().map(modelB -> {
            storeInDB()l // store B here ?
            return modelB;
        });
    });

If I wasn't using lambda functions, this would look as ugly as normal nested calls. Is this a better way to do it?

Solution 1

I don't think using map operator is the best way to go with things like storing the result of the api call.

What I like to do is to separate those things inside doOnNext operators. So your example would be something like this:

apiService.A()
        .doOnNext(modelA -> db.store(modelA))
        .flatMap(modelA -> apiService.B())
        .doOnNext(modelB -> db.store(modelB));

(add necessary observeOn and subscribeOn yourself, exactly like you need them)

Solution 2

Yes, you can use flatmap for this exact purpose. See the below example (Assuming your service A returns Observable<FooA> and service B returns Observable<FooB>)

api.serviceA()
        .flatMap(new Func1<FooA, Observable<FooB>>() {
            @Override
            public Observable<FooB> call(FooA fooA) {
                // code to save data from service A to db

                // call service B
                return api.serviceB();
            }
         })
         .subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread())
         .subscribe(new Subscriber<FooB>() {
             @Override
             public void onCompleted() {
             }

             @Override
             public void onError(Throwable e) {

             }

             @Override
             public void onNext(FooB fooB) {
                 // code to save data from service B to db

             }
        });

Comments

[Sourabh]

I didn't know about do operators of Rx. After reading its docs, isn't doOnComplete a better function for this?

[Bartek Lipinski]

@Sourabh it depends on your Observable. If onComplete is called after a successful call from api, then yes. If your Observable can emit multiple events, then you should go for doOnNext. In your case (Retrofit) using doOnNext and doOnComplete will have the same result, but indeed doOnComplete might be a bit more intuitive (I used doOnNext out of pure habit).

[Sourabh]

I meant, use doOnComplete instead of flatMap

[Sourabh]

... as B() doesn't depend on result of A() (SO's 5 min comment edit time)

[Bartek Lipinski]

using doOnComplete instead of flatMap would be wrong, because doOnComplete wouldn't change the Observable in stream, and flatMap would. In other words, if you use flatMap you can use just one subscribe, and trying to use doOnComplete instead of this would force you to call apiService.B().subscribe(...) inside this doOnComplete (which would look ugly as hell)

[Sudheesh Mohan]

If I write code to save data to DB in onNext it will run on main thread right? How to save data to db on background thread instead?

[Much Overflow]

throw in another flatMap operator. api.serviceA().flatMap((FooA) -> {// call service B}).flatMap((FooB) -> {// save data from service B and return a observable stating success / failure}).subscribeOn(Schedulers.io()).observeOn(AndroidSche‌​dulers.mainThread())‌​.doOnNext((isSuccess‌​) -> {// Only this will execute on main thread})

[Ryhan]

How would you do this in RxJava2?

[Andrii Kovalchuk]

Use this for a Single: .doOnEvent { t1, t2 -> }