@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")); }
static <F, E> ObservableTransformer<F, E> fromConsumer(final Consumer<F> doEffect) { return fromConsumer(doEffect, null); }
Transformers { static <F, E> ObservableTransformer<F, E> fromConsumer(final Consumer<F> doEffect) { return fromConsumer(doEffect, null); } }
Transformers { static <F, E> ObservableTransformer<F, E> fromConsumer(final Consumer<F> doEffect) { return fromConsumer(doEffect, null); } private  Transformers(); }
Transformers { static <F, E> ObservableTransformer<F, E> fromConsumer(final Consumer<F> doEffect) { return fromConsumer(doEffect, null); } private  Transformers();  }
Transformers { static <F, E> ObservableTransformer<F, E> fromConsumer(final Consumer<F> doEffect) { return fromConsumer(doEffect, null); } private  Transformers();   }