How should I zip a List> into an Observable>?

Question

I have a List<Observable<T>> that I would like to transform into an Observable<List<T>>.

I am aware of Observable.zip, which seems like the right function, but I'm not sure how to define the zipper parameter.

Here is what I have tried:

final List<Observable<T>> tasks = getTasks();

final Observable<List<T>> task = Observable.zip(
    tasks, 
    x -> ImmutableList.copyOf(x)
        .stream()
        .map(x -> (T)x)
        .collect(ImmutableList.toImmutableList()));

However, this requires an unchecked cast.

How should I go about this in RxJava 2?

Note that this question refers to RxJava 1.

score 1 · Accepted Answer · answered May 24 '17 at 20:52

1

Use the primitives:

List<Observable<T>> tasks = getTasks();
Observable<List<T>> task = Observable.merge(tasks).toList();

However, do you really need all the tasks at once? You could skip the toList and proccess the tasks as they come; this will give you both better responsiveness and easier concurrency control.

answered May 24 '17 at 20:52

Tassos Bassoukos

16,017
2
36
40

I have a list of files to download and I want to wait for all to succeed or one to fail. – sdgfsdh May 24 '17 at 20:53
Then you don't need `toList` - process them one-by-one, and the `merge` operator will unsubscribe from all source observables if one of them emits an error. – Tassos Bassoukos May 24 '17 at 20:54
Just make sure that your process can handle unsubscription. – Tassos Bassoukos May 24 '17 at 20:55
What about updates? If I make an `Observable>` then I can read the progress of each download over time. – sdgfsdh May 24 '17 at 20:56

score 0 · Answer 2 · answered May 24 '17 at 20:45

Something like this should do

final List<Observable<T>> tasks = getTasks();
List<T> values = getTasks().stream()
                           .map(observable -> observable.firstElement().blockingGet())
                           .collect(Collectors.toList());
Observable<List<T>> result = Observable.fromArray(values);

How should I zip a List> into an Observable>?

2 Answers2