Transformers { static <F, E> ObservableTransformer<F, E> fromConsumer(final Consumer<F> doEffect) { return fromConsumer(doEffect, null); } private  Transformers();   }
@Test public void effectPerformerInvokesConsumerAndPassesTheRequestedEffect() throws Exception { PublishSubject<String> upstream = PublishSubject.create(); TestConsumer<String> consumer = new TestConsumer<>(); upstream.compose(Transformers.fromConsumer(consumer)).subscribe(); upstream.onNext("First Time"); assertThat(consumer.getCurrentValue(), is("First Time")); upstream.onNext("Do it again!"); assertThat(consumer.getCurrentValue(), is("Do it again!")); }
@Test public void effectPerformerInvokesConsumerOnSchedulerAndPassesTheRequestedEffect() throws Exception { PublishSubject<String> upstream = PublishSubject.create(); TestConsumer<String> consumer = new TestConsumer<>(); TestScheduler scheduler = new TestScheduler(); upstream.compose(Transformers.fromConsumer(consumer, scheduler)).subscribe(); upstream.onNext("First Time"); assertThat(consumer.getCurrentValue(), is(equalTo(null))); scheduler.triggerActions(); assertThat(consumer.getCurrentValue(), is("First Time")); }