如何组合最近的一个Observable <Observable <T>>？

Question

我有一个Observable<Observable<String>> c。我想用新行将每个内部Observable合并为一个String。结果应该是Observable<String>。

有一个similar discussion for RxJS on GitHub。

下面是使用zip一个实现：

final Observable<String> d = Observable.zip(c, objects -> Arrays.stream(objects) 
    .map(x -> (String)x) 
    .collect(Collectors.joining("\n"))); 

的问题是，zip对元素，因此如果内Observable对象具有不同的长度，那么不一切都被示出。我想用combineLatest代替，但似乎没有成为一个匹配功能：

public static <T, R> Observable<R> combineLatest(
    ObservableSource<? extends ObservableSource<? extends T>> sources, 
    Function<? super Object[], ? extends R> zipper) 

我应该如何实现呢？

---

举例来说，如果我有三个观测：

// a 
Observable.just("Started", "0%", "5%", "10%", "20%", "40%", "95%", "100%", "Finished") 

// b 
Observable.just("Started", "0%", "10%", "30%", "40%", "100%", "Finished") 

// c 
Observable.just("Started", "0%", "20%", "25%", "40%", "70%", "90%", "100%", "Finished") 

然后我zip解决方案的输出是：

--------------------- 
a: Started 
b: Started 
c: Started 
--------------------- 
a: 0% 
b: 0% 
c: 0% 
--------------------- 
a: 5% 
b: 10% 
c: 20% 
--------------------- 
a: 10% 
b: 30% 
c: 25% 
--------------------- 
a: 20% 
b: 40% 
c: 40% 
--------------------- 
a: 40% 
b: 100% 
c: 70% 
--------------------- 
a: 95% 
b: Finished 
c: 90% 

但所需的输出是一样的东西：

--------------------- 
a: Started 
b: Started 
c: Started 
--------------------- 
a: 0% 
b: 0% 
c: 0% 
--------------------- 
a: 5% 
b: 10% 
c: 20% 
--------------------- 
a: 10% 
b: 30% 
c: 25% 
--------------------- 
a: 20% 
b: 40% 
c: 40% 
--------------------- 
a: 40% 
b: 100% 
c: 70% 
--------------------- 
a: 95% 
b: Finished 
c: 90% 
--------------------- 
a: 100% 
b: Finished 
c: 100% 
--------------------- 
a: Finished 
b: Finished 
c: Finished 

Answer 1

使用RxJava2（和Java 8）

c.toList().toObservable() 
      .flatMap(t -> 
        Observable.combineLatest(t, 
              objects -> Arrays.stream(objects) 
                  .map(x ->(String)x).collect(Collectors.joining("\n")))) 

使用RxJava

c.toList().flatMap(t -> 
        Observable.combineLatest(t, 
              objects -> Arrays.stream(objects) 
                  .map(x -> (String)x).collect(Collectors.joining("\n")))); 

我怀疑它会做你想要什么，但因为它是完全有可能将“跳过”排放量，但就是这样combineLatest作品。