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);  }
@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()); }
@Test public void testNumberOfThreads() { final ConcurrentHashMap<Long, Long> threads = new ConcurrentHashMap<Long, Long>(); OperationParallelMerge.parallelMerge(getStreams(), 3) .flatMap(new Func1<Observable<String>, Observable<String>>() { @Override public Observable<String> call(Observable<String> o) { return o.observeOn(Schedulers.newThread()); } }) .toBlockingObservable().forEach(new Action1<String>() { @Override public void call(String o) { System.out.println("o: " + o + " Thread: " + Thread.currentThread().getId()); threads.put(Thread.currentThread().getId(), Thread.currentThread().getId()); } }); assertEquals(3, threads.keySet().size()); }
@Test public void testNumberOfThreadsOnScheduledMerge() { final ConcurrentHashMap<Long, Long> threads = new ConcurrentHashMap<Long, Long>(); Observable.merge(OperationParallelMerge.parallelMerge(getStreams(), 3, Schedulers.newThread())) .toBlockingObservable().forEach(new Action1<String>() { @Override public void call(String o) { System.out.println("o: " + o + " Thread: " + Thread.currentThread().getId()); threads.put(Thread.currentThread().getId(), Thread.currentThread().getId()); } }); assertEquals(3, threads.keySet().size()); }