RxJava - parallel execution of two API calls

Question

I have a SyncService.

@Override
public int onStartCommand(Intent intent, int flags, final int startId) {
    Timber.i("Starting sync...");
    ...
    RxUtil.unsubscribe(mSubscription);
    mSubscription = mDataManager.syncEvents()
            .subscribeOn(Schedulers.io())
            .subscribe(new Observer<Event>() {
                @Override
                public void onCompleted() {
                    Timber.i("Synced successfully!");
                    stopSelf(startId);
                }

                @Override
                public void onError(Throwable e) {
                    Timber.w(e, "Error syncing.");
                    stopSelf(startId);

                }

                @Override
                public void onNext(Event event) {
                }
            });

    return START_STICKY;
}

Observable events = mDataManager.syncEvents() is an API call.

I want to do a parallel call:

Single userInfo = mDataManager.getUserInfo()

and call stopSelf(startId); after these two calls will finish.

How can I do it?

I tried RxJava Fetching Observables In Parallel but this is a little different case.

I think I have to use .zip or .merge method. But in my case one method call returns Observable (list on Events) and second Single (one UserInfo object).

I created z result class which could be a result of .zip method, but I don't know how to fill it:

public class SyncResponse {
     List<Event> events;
     UserInfo userInfo;
     ...
}

Anton Pogonets Anton Pogonets · Accepted Answer · 2016-12-27T21:45:38

Since you have just two observables you can combine them using android Pair to combine results.

mDataManager.syncEvents()
    .zipWith(mDataManager.getUserInfo().toObservable(), Pair::create)
    .subscribeOn(Schedulers.io())
    .subscribe(pair -> {
        Event event = pair.first;
        UserInfo userInfo = pair.second;

        // your code here

        Timber.i("Synced successfully!");
        stopSelf(startId);
    }, throwable -> {
        Timber.w(e, "Error syncing.");
        stopSelf(startId);
    });

You can do the same without java8/retrolmbda but why

If you need collect all events till completed and combine them all with single user info it would be little bit complicated

Observable<ArrayList<String>> eventsObservable = mDataManager.syncEvents()
    .collect(ArrayList::new, ArrayList::add); 

mDataManager.getUserInfo()
    .zipWith(eventsObservable.toSingle(), Pair::create)
    .subscribeOn(Schedulers.io())
    .subscribe(pair -> {
        UserInfo userInfo = pair.first;
        List<Event> event = pair.second;

        // your code here

        Timber.i("Synced successfully!");
        stopSelf(startId);
    }, throwable -> {
        Timber.w(e, "Error syncing.");
        stopSelf(startId);
    });

RxJava - parallel execution of two API calls

2 Answers