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