Transformers { static <F, E> Transformer<F, E> fromConsumer(final Action1<F> doEffect) { return fromConsumer(doEffect, null); } private  Transformers();   }
@Test public void effectPerformerInvokesConsumerAndPassesTheRequestedEffect() throws Exception { 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 { 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")); }
@Test public void consumerTransformerShouldPropagateCompletion() throws Exception { AssertableSubscriber<Object> subscriber = upstream.compose(Transformers.fromConsumer(consumer, scheduler)).test(); upstream.onNext("hi"); upstream.onCompleted(); scheduler.triggerActions(); subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); subscriber.assertCompleted(); }