@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()); }
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);  }