"Join" a list of Observables in RxJava

Question

I have been reading the documentation over and over again and I cannot find the right operator for my case:

 return Observable.fromIterable(formIds)
                            .flatMap(new Function<String, Observable<List<Transmission>>>() {
                                @Override
                                public Observable<List<Transmission>> apply(String formId) {
                                    return getFormTransmissions(formId);
                                }
                            })
                            .toList()
                            .toObservable()
                            .map(new Function<List<List<Transmission>>, List<Transmission>>() {
                                @Override
                                public List<Transmission> apply(List<List<Transmission>> lists) {
                                    List<Transmission> transmissions = new ArrayList<>();
                                    for (List<Transmission> transmissionList : lists) {
                                        if (transmissionList != null) {
                                            transmissions.addAll(transmissionList);
                                        }
                                    }
                                    return transmissions;
                                }
                            });

As you can see the last map() I take all the observables and combine their content. I don't like this part as I feel there must be some rxjava function to do that but I cannot find one.

I will try to explain what the code does: From a list of ids For each one of them, get a list of transmissions for each id Once all of those complete combine all the individual transmission lists into one. Any ideas?

Answer 1

So basically you need a simple map that can flatten the result. You can do this with flatMapIterable

return Observable.fromIterable(formIds)
        .flatMap(new Function<String, Observable<List<Transmission>>>() {
            @Override
            public Observable<List<Transmission>> apply(String formId) {
                return getFormTransmissions(formId);
            }
        }) // Observable<List<Transmission>>
        .flatMapIterable(new Function<List<Transmission>, List<Transmission>>() {
                @Override
                public List<Transmission> apply(List<Transmission> list) {
                    return list;
                }
            }) // Observable<Transmission>
        .toList() // Single<List<Transmission>>
        .toObservable(); // Observable<List<Transmission>>

Even simpler with Java 8 features:

return Observable.fromIterable(formIds)
        .flatMap(formId -> getFormTransmissions(formId))  // Observable<List<Transmission>>
        .flatMapIterable(list -> list) // Observable<Transmission>
        .toList() // Single<List<Transmission>>
        .toObservable(); // Observable<List<Transmission>>

Alternatively you can just use flatMap in Java Stream:

return Observable.fromIterable(formIds)
        .flatMap(formId -> getFormTransmissions(formId))  // Observable<List<Transmission>>
        .toList() // Single<List<List<Transmission>>>
        .toObservable() // Observable<List<List<Transmission>>>
        .map(listOfList -> listOfList.stream().flatMap(Collection::stream)
                .collect(Collectors.toList()));  // Observable<List<Transmission>>