@Test public void testParallelMerge() { PublishSubject<String> p1 = PublishSubject.<String> create(); PublishSubject<String> p2 = PublishSubject.<String> create(); PublishSubject<String> p3 = PublishSubject.<String> create(); PublishSubject<String> p4 = PublishSubject.<String> create(); Observable<Observable<String>> fourStreams = Observable.<Observable<String>> from(p1, p2, p3, p4); Observable<Observable<String>> twoStreams = OperationParallelMerge.parallelMerge(fourStreams, 2); Observable<Observable<String>> threeStreams = OperationParallelMerge.parallelMerge(fourStreams, 3); List<? super Observable<String>> fourList = fourStreams.toList().toBlockingObservable().last(); List<? super Observable<String>> threeList = threeStreams.toList().toBlockingObservable().last(); List<? super Observable<String>> twoList = twoStreams.toList().toBlockingObservable().last(); System.out.println("two list: " + twoList); System.out.println("three list: " + threeList); System.out.println("four list: " + fourList); assertEquals(4, fourList.size()); assertEquals(3, threeList.size()); assertEquals(2, twoList.size()); }
public static <T> Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables) { return parallelMerge(source, parallelObservables, Schedulers.currentThread()); }
OperationParallelMerge { public static <T> Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables) { return parallelMerge(source, parallelObservables, Schedulers.currentThread()); } }
OperationParallelMerge { public static <T> Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables) { return parallelMerge(source, parallelObservables, Schedulers.currentThread()); }  }
OperationParallelMerge { public static <T> Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables) { return parallelMerge(source, parallelObservables, Schedulers.currentThread()); }  static Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables); static Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables, final Scheduler scheduler); }
OperationParallelMerge { public static <T> Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables) { return parallelMerge(source, parallelObservables, Schedulers.currentThread()); }  static Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables); static Observable<Observable<T>> parallelMerge(final Observable<Observable<T>> source, final int parallelObservables, final Scheduler scheduler);  }