@Test public void create() throws ExecutionException, InterruptedException { Multi<Integer> publisher = Multi.createFrom() .items(1, 2, 1, 4, 3, 2, 5, 3, 6, 6, 4, 8, 7, 4, 5, 6, 7, 8, 9, 9, 10) .emitOn(executor); List<String> list = ReactiveStreams.fromPublisher(publisher) .filter(i -> i < 4) .distinct() .map(this::asString) .toList() .run().toCompletableFuture().get(); assertThat(list).containsExactly("1", "2", "3"); }
@SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage) { Objects.requireNonNull(stage); return source -> (Multi<O>) source.transform().byDroppingDuplicates(); }
DistinctStageFactory implements ProcessingStageFactory<Stage.Distinct> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage) { Objects.requireNonNull(stage); return source -> (Multi<O>) source.transform().byDroppingDuplicates(); } }
DistinctStageFactory implements ProcessingStageFactory<Stage.Distinct> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage) { Objects.requireNonNull(stage); return source -> (Multi<O>) source.transform().byDroppingDuplicates(); }  }
DistinctStageFactory implements ProcessingStageFactory<Stage.Distinct> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage) { Objects.requireNonNull(stage); return source -> (Multi<O>) source.transform().byDroppingDuplicates(); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage); }
DistinctStageFactory implements ProcessingStageFactory<Stage.Distinct> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage) { Objects.requireNonNull(stage); return source -> (Multi<O>) source.transform().byDroppingDuplicates(); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Distinct stage);  }