### Question:
FromPublisherStageFactory implements PublisherStageFactory<Stage.PublisherStage> { @SuppressWarnings("unchecked") @Override public <O> PublisherStage<O> create(Engine engine, Stage.PublisherStage stage) { Publisher<O> publisher = (Publisher<O>) Objects.requireNonNull(Objects.requireNonNull(stage.getRsPublisher())); return () -> Multi.createFrom().publisher(publisher); }  @SuppressWarnings("unchecked") @Override PublisherStage<O> create(Engine engine, Stage.PublisherStage stage);  }### Answer:
@Test public void create() throws ExecutionException, InterruptedException { List<Integer> list = ReactiveStreams.fromPublisher(Multi.createFrom().items(1, 2, 3)).toList().run() .toCompletableFuture() .get(); assertThat(list).containsExactly(1, 2, 3); Optional<Integer> res = ReactiveStreams.fromPublisher(Multi.createFrom().item(25)).findFirst().run() .toCompletableFuture() .get(); assertThat(res).contains(25); Optional<?> empty = ReactiveStreams.fromPublisher(Multi.createFrom().empty()).findFirst().run() .toCompletableFuture() .get(); assertThat(empty).isEmpty(); try { ReactiveStreams.fromPublisher(Multi.createFrom().failure(new Exception("Boom"))).findFirst().run() .toCompletableFuture().get(); fail("Exception should be thrown"); } catch (Exception e) { assertThat(e).hasMessageContaining("Boom"); } }
@SuppressWarnings("ConstantConditions") @Test public void createWithoutStage() { assertThrows(NullPointerException.class, () -> factory.create(null, null)); }
@Test public void createWithoutFunction() { assertThrows(NullPointerException.class, () -> factory.create(null, () -> null)); }