RxJava Combine Sequence Of Requests

the flatMap operator is designed to cater to these types of workflows.

i'll outline the broad strokes with a simple five step example. hopefully you can easily reconstruct the same principles in your code:

@Test fun flatMapExample() {
    // (1) constructing a fake stream that emits a list of values
    Observable.just(listOf(1, 2, 3, 4, 5))
            // (2) convert our List emission into a stream of its constituent values 
            .flatMap { numbers -> Observable.fromIterable(numbers) }
            // (3) subsequently convert each individual value emission into an Observable of some 
            //     newly calculated type
            .flatMap { number ->
                when(number) {
                       1 -> Observable.just("A1")
                       2 -> Observable.just("B2")
                       3 -> Observable.just("C3")
                       4 -> Observable.just("D4")
                       5 -> Observable.just("E5")
                    else -> throw RuntimeException("Unexpected value for number [$number]")
                }
            }
            // (4) collect all the final emissions into a list
            .toList()
            .subscribeBy(
                    onSuccess = {
                        // (5) handle all the combined results (in list form) here
                        println("## onNext($it)")
                    },
                    onError = { error ->
                        println("## onError(${error.message})")
                    }
            )
}

(incidentally, if the order of the emissions matter, look at using concatMap instead).

i hope that helps.

Check below it's working.

Say you have multiple network calls you need to make–cals to get Github user information and Github user events for example.

And you want to wait for each to return before updating the UI. RxJava can help you here. Let’s first define our Retrofit object to access Github’s API, then setup two observables for the two network requests call.

Retrofit repo = new Retrofit.Builder()
        .baseUrl("https://api.github.com")
        .addConverterFactory(GsonConverterFactory.create())
        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
        .build();

Observable<JsonObject> userObservable = repo
        .create(GitHubUser.class)
        .getUser(loginName)
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread());

Observable<JsonArray> eventsObservable = repo
        .create(GitHubEvents.class)
        .listEvents(loginName)
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread());

Used Interface for it like below:

public interface GitHubUser {
  @GET("users/{user}")
  Observable<JsonObject> getUser(@Path("user") String user);
}

public interface GitHubEvents {
  @GET("users/{user}/events")
  Observable<JsonArray> listEvents(@Path("user") String user);
}

After we use RxJava’s zip method to combine our two Observables and wait for them to complete before creating a new Observable.

Observable<UserAndEvents> combined = Observable.zip(userObservable, eventsObservable, new Func2<JsonObject, JsonArray, UserAndEvents>() {
  @Override
  public UserAndEvents call(JsonObject jsonObject, JsonArray jsonElements) {
    return new UserAndEvents(jsonObject, jsonElements);
  }
});

Finally let’s call the subscribe method on our new combined Observable:

combined.subscribe(new Subscriber<UserAndEvents>() {
          ...
          @Override
          public void onNext(UserAndEvents o) {
            // You can access the results of the 
            // two observabes via the POJO now
          }
        });

No more waiting in threads etc for network calls to finish. RxJava has done all that for you in zip(). hope my answer helps you.

I solved a similar problem with RxJava2. Execution of requests for Api 2 in parallel slightly speeds up the work.

private InformationRepository informationRepository;

//init....

public Single<List<FullInformation>> getFullInformation() {
    return informationRepository.getInformationList()
            .subscribeOn(Schedulers.io())//I usually write subscribeOn() in the repository, here - for clarity
            .flatMapObservable(Observable::fromIterable)
            .flatMapSingle(this::getFullInformation)
            .collect(ArrayList::new, List::add);

}

private Single<FullInformation> getFullInformation(Information information) {
    return informationRepository.getExtendedInformation(information)
            .map(extendedInformation -> new FullInformation(information, extendedInformation))
            .subscribeOn(Schedulers.io());//execute requests in parallel
}

InformationRepository - just interface. Its implementation is not interesting for us.

public interface InformationRepository {

    Single<List<Information>> getInformationList();

    Single<ExtendedInformation> getExtendedInformation(Information information);
}

FullInformation - container for result.

public class FullInformation {

    private Information information;
    private ExtendedInformation extendedInformation;

    public FullInformation(Information information, ExtendedInformation extendedInformation) {
        this.information = information;
        this.extendedInformation = extendedInformation;
    }
}

Try using Observable.zip() operator. It will wait until both Api calls are finished before continuing the stream. Then you can insert some logic by calling flatMap() afterwards.

http://reactivex.io/documentation/operators/zip.html