Transformers { static <F, E> ObservableTransformer<F, E> fromAction(final Action doEffect) { return fromAction(doEffect, null); } private  Transformers();   }
@Test public void effectPerformerRunsActionWheneverEffectIsRequested() throws Exception { PublishSubject<String> upstream = PublishSubject.create(); TestAction action = new TestAction(); upstream.compose(Transformers.fromAction(action)).subscribe(); upstream.onNext("First Time"); assertThat(action.getRunCount(), is(1)); upstream.onNext("One more!"); assertThat(action.getRunCount(), is(2)); }
@Test public void effectPerformerRunsActionOnSchedulerWheneverEffectIsRequested() throws Exception { PublishSubject<String> upstream = PublishSubject.create(); TestAction action = new TestAction(); TestScheduler scheduler = new TestScheduler(); upstream.compose(Transformers.fromAction(action, scheduler)).subscribe(); upstream.onNext("First Time"); assertThat(action.getRunCount(), is(0)); scheduler.triggerActions(); assertThat(action.getRunCount(), is(1)); }