### Question:
SubscriberStageFactory implements TerminalStageFactory<Stage.SubscriberStage> { @SuppressWarnings("unchecked") @Override public <I, O> TerminalStage<I, O> create(Engine engine, Stage.SubscriberStage stage) { Subscriber<I> subscriber = (Subscriber<I>) Objects.requireNonNull(stage).getRsSubscriber(); Objects.requireNonNull(subscriber); return (TerminalStage<I, O>) new SubscriberStage<>(subscriber); }  @SuppressWarnings("unchecked") @Override TerminalStage<I, O> create(Engine engine, Stage.SubscriberStage stage);  }### Answer:
@Test public void create() { Multi<Integer> publisher = Multi.createFrom().items(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .emitOn(executor); SubscriberBuilder<Integer, Optional<Integer>> builder = ReactiveStreams.<Integer> builder().findFirst(); Optional<Integer> optional = ReactiveStreams.fromPublisher(publisher).filter(i -> i > 5) .to(builder).run().toCompletableFuture().join(); assertThat(optional).contains(6); }
@Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> factory.create(null, null)); }
@Test public void createWithoutSubscriber() { assertThrows(NullPointerException.class, () -> factory.create(new Engine(), () -> null)); }