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 () -> Flowable.fromPublisher(publisher); }  @SuppressWarnings("unchecked") @Override PublisherStage<O> create(Engine engine, Stage.PublisherStage stage);  }
@Test public void create() throws ExecutionException, InterruptedException { List<Integer> list = ReactiveStreams.fromPublisher(Flowable.fromArray(1, 2, 3)).toList().run().toCompletableFuture() .get(); assertThat(list).containsExactly(1, 2, 3); Optional<Integer> res = ReactiveStreams.fromPublisher(Flowable.just(25)).findFirst().run().toCompletableFuture().get(); assertThat(res).contains(25); Optional<?> empty = ReactiveStreams.fromPublisher(Flowable.empty()).findFirst().run().toCompletableFuture().get(); assertThat(empty).isEmpty(); try { ReactiveStreams.fromPublisher(Flowable.error(new Exception("Boom"))).findFirst().run() .toCompletableFuture().get(); fail("Exception should be thrown"); } catch (Exception e) { assertThat(e).hasMessageContaining("Boom"); } }
@Test(expected = NullPointerException.class) public void createWithoutStage() { factory.create(null, null); }
@Test(expected = NullPointerException.class) public void createWithoutFunction() { factory.create(null, () -> null); }