class 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);
    }

}