### Question:
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 -> (Multi<O>) source .onItem().invoke(consumer); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Peek stage);  }### Answer:
@Test public void create() throws ExecutionException, InterruptedException { Multi<Integer> publisher = Multi.createFrom().items(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .emitOn(executor); List<Integer> squares = new ArrayList<>(); List<String> strings = new ArrayList<>(); List<String> list = ReactiveStreams.fromPublisher(publisher) .filter(i -> i < 4) .map(this::square) .peek(squares::add) .map(this::asString) .peek(strings::add) .toList() .run().toCompletableFuture().get(); assertThat(list).containsExactly("1", "4", "9"); assertThat(squares).containsExactly(1, 4, 9); assertThat(strings).containsExactly("1", "4", "9"); }
@Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> factory.create(null, null)); }
@Test public void createWithoutFunction() { assertThrows(NullPointerException.class, () -> factory.create(null, () -> null)); }