@Test(expected = NullPointerException.class) public void createWithoutFunction() { factory.create(null, () -> null); }
@SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Peek stage) { Consumer<I> consumer = (Consumer<I>) Objects.requireNonNull(stage) .getConsumer(); Objects.requireNonNull(consumer); return source -> (Flowable<O>) source.doOnNext(consumer::accept); }
PeekStageFactory implements ProcessingStageFactory<Stage.Peek> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Peek stage) { Consumer<I> consumer = (Consumer<I>) Objects.requireNonNull(stage) .getConsumer(); Objects.requireNonNull(consumer); return source -> (Flowable<O>) source.doOnNext(consumer::accept); } }
PeekStageFactory implements ProcessingStageFactory<Stage.Peek> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Peek stage) { Consumer<I> consumer = (Consumer<I>) Objects.requireNonNull(stage) .getConsumer(); Objects.requireNonNull(consumer); return source -> (Flowable<O>) source.doOnNext(consumer::accept); }  }
PeekStageFactory implements ProcessingStageFactory<Stage.Peek> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Peek stage) { Consumer<I> consumer = (Consumer<I>) Objects.requireNonNull(stage) .getConsumer(); Objects.requireNonNull(consumer); return source -> (Flowable<O>) source.doOnNext(consumer::accept); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Peek stage); }
PeekStageFactory implements ProcessingStageFactory<Stage.Peek> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Peek stage) { Consumer<I> consumer = (Consumer<I>) Objects.requireNonNull(stage) .getConsumer(); Objects.requireNonNull(consumer); return source -> (Flowable<O>) source.doOnNext(consumer::accept); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Peek stage);  }