@Test public void processingLongEffectsDoesNotBlockProcessingShorterEffects() { final List<String> effects = Arrays.asList("Hello", "Rx"); PublishSubject<String> upstream = PublishSubject.create(); Function<String, Integer> sleepyFunction = s -> { try { Thread.sleep(duration(s)); } catch (InterruptedException ie) { } return s.length(); }; final List<Integer> results = new ArrayList<>(); final Disposable disposable = upstream .compose(Transformers.fromFunction(sleepyFunction, Schedulers.io())) .subscribe(results::add); Observable.fromIterable(effects).subscribe(upstream); await().atMost(durationForEffects(effects)).until(() -> results.equals(expected(effects))); disposable.dispose(); }
static <F, E> ObservableTransformer<F, E> fromFunction( final Function<F, E> function, @Nullable final Scheduler scheduler) { return new ObservableTransformer<F, E>() { @Override public ObservableSource<E> apply(Observable<F> effectStream) { return effectStream.flatMap( new Function<F, ObservableSource<E>>() { @Override public ObservableSource<E> apply(@NonNull F f) { Observable<E> eventObservable = Observable.fromSupplier( new Supplier<E>() { @Override public E get() throws Throwable { return function.apply(f); } }); return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler); } }); } }; }
Transformers { static <F, E> ObservableTransformer<F, E> fromFunction( final Function<F, E> function, @Nullable final Scheduler scheduler) { return new ObservableTransformer<F, E>() { @Override public ObservableSource<E> apply(Observable<F> effectStream) { return effectStream.flatMap( new Function<F, ObservableSource<E>>() { @Override public ObservableSource<E> apply(@NonNull F f) { Observable<E> eventObservable = Observable.fromSupplier( new Supplier<E>() { @Override public E get() throws Throwable { return function.apply(f); } }); return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler); } }); } }; } }
Transformers { static <F, E> ObservableTransformer<F, E> fromFunction( final Function<F, E> function, @Nullable final Scheduler scheduler) { return new ObservableTransformer<F, E>() { @Override public ObservableSource<E> apply(Observable<F> effectStream) { return effectStream.flatMap( new Function<F, ObservableSource<E>>() { @Override public ObservableSource<E> apply(@NonNull F f) { Observable<E> eventObservable = Observable.fromSupplier( new Supplier<E>() { @Override public E get() throws Throwable { return function.apply(f); } }); return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler); } }); } }; } private  Transformers(); }
Transformers { static <F, E> ObservableTransformer<F, E> fromFunction( final Function<F, E> function, @Nullable final Scheduler scheduler) { return new ObservableTransformer<F, E>() { @Override public ObservableSource<E> apply(Observable<F> effectStream) { return effectStream.flatMap( new Function<F, ObservableSource<E>>() { @Override public ObservableSource<E> apply(@NonNull F f) { Observable<E> eventObservable = Observable.fromSupplier( new Supplier<E>() { @Override public E get() throws Throwable { return function.apply(f); } }); return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler); } }); } }; } private  Transformers();  }
Transformers { static <F, E> ObservableTransformer<F, E> fromFunction( final Function<F, E> function, @Nullable final Scheduler scheduler) { return new ObservableTransformer<F, E>() { @Override public ObservableSource<E> apply(Observable<F> effectStream) { return effectStream.flatMap( new Function<F, ObservableSource<E>>() { @Override public ObservableSource<E> apply(@NonNull F f) { Observable<E> eventObservable = Observable.fromSupplier( new Supplier<E>() { @Override public E get() throws Throwable { return function.apply(f); } }); return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler); } }); } }; } private  Transformers();   }