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(final F f) { Observable<E> eventObservable = Observable.fromCallable( new Callable<E>() { @Override public E call() throws Exception { return function.apply(f); } }); return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler); } }); } }; } private  Transformers();   }
@Test public void effectPerformerInvokesFunctionWithReceivedEffectAndErrorsForUnhandledExceptions() { PublishSubject<String> upstream = PublishSubject.create(); TestScheduler scheduler = new TestScheduler(); Function<String, Integer> function = s -> { throw new RuntimeException("Something bad happened"); }; TestObserver<Integer> observer = upstream.compose(Transformers.fromFunction(function, scheduler)).test(); upstream.onNext("Hello"); scheduler.triggerActions(); observer.assertError(RuntimeException.class); }
@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(); }
@Test public void effectPerformerInvokesFunctionWithReceivedEffectAndEmitsReturnedEvents() { PublishSubject<String> upstream = PublishSubject.create(); TestScheduler scheduler = new TestScheduler(); Function<String, Integer> function = s -> s.length(); TestObserver<Integer> observer = upstream.compose(Transformers.fromFunction(function, scheduler)).test(); upstream.onNext("Hello"); scheduler.triggerActions(); observer.assertValue(5); }