@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> list = ReactiveStreams.fromPublisher(publisher).limit(5).toList().run() .toCompletableFuture().get(); assertThat(list).hasSize(5).containsExactly(1, 2, 3, 4, 5); }
@SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Limit stage) { long limit = stage.getLimit(); return source -> (Multi<O>) source.transform().byTakingFirstItems(limit); }
LimitStageFactory implements ProcessingStageFactory<Stage.Limit> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Limit stage) { long limit = stage.getLimit(); return source -> (Multi<O>) source.transform().byTakingFirstItems(limit); } }
LimitStageFactory implements ProcessingStageFactory<Stage.Limit> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Limit stage) { long limit = stage.getLimit(); return source -> (Multi<O>) source.transform().byTakingFirstItems(limit); }  }
LimitStageFactory implements ProcessingStageFactory<Stage.Limit> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Limit stage) { long limit = stage.getLimit(); return source -> (Multi<O>) source.transform().byTakingFirstItems(limit); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Limit stage); }
LimitStageFactory implements ProcessingStageFactory<Stage.Limit> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Limit stage) { long limit = stage.getLimit(); return source -> (Multi<O>) source.transform().byTakingFirstItems(limit); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Limit stage);  }