RxJava: динамически создавать Observables и отправлять окончательный результат как Observable

Question

Я использую RxJava, в котором хочу динамически создавать несколько Observables на основе некоторого условия. Как только я закончил создание, я хочу сделать некоторую обработку на разных значениях, возвращаемых наблюдаемыми, а затем отправить как единый Observable, на который я могу подписаться. Вот как мой код:

List<String> valueList = .... 

List<Observable<String>> listOfObservables = new ArrayList<Observable<String>>(); 

for(int i =; i <valueList.size(); i++){ 
     listOfObservables.add(new SomeClass.doOperation(valueList(i))); 
     // SomeClass.doOperation will return an Observable<String> 
    } 

return Observable.merge(listOfObservables); 

Но здесь, я хочу сделать некоторые операции по значениям, излучаемых различными наблюдаемыми, в listOfObservable и, наконец, вернуть его в качестве одного Observable<String>

Как и в Observable.zip(), я может сделать это как

return Observable.zip(observable1, observable2, (string1, string2) -> { 
      // joining final string here 
      return string1 + string2; 

Но я знаю количество аргументов здесь. Пожалуйста, дайте мне знать, как я могу это достичь.

Answer 1

Используйте zip перегрузки that takes a variable number of arguments, она имеет подпись использования

<R> Observable<R> zip(Iterable<? extends Observable<?>> ws, 
         FuncN<? extends R> zipFunction) 

Пример:

List<String> valueList = .... 

return Observable.from(valueList) 
    .map(string -> SomeClass.doOperationThatReturnsObservable(string)) 
    .toList() 
    .flatMap(listOfObs -> Observable.zip(listOfObs, (Object[] results) -> { 
     // do something with the strings in the array. 
     return Arrays.stream(results) 
        .map(Object::toString) 
        .collect(Collectors.joining(",")); 
    }));