@Test public void createWithoutPredicate() { assertThrows(NullPointerException.class, () -> factory.create(null, () -> null)); }
@SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Filter stage) { Objects.requireNonNull(stage); Predicate predicate = Objects.requireNonNull(stage.getPredicate()); return source -> (Multi<O>) source.transform().byFilteringItemsWith(predicate); }
FilterStageFactory implements ProcessingStageFactory<Stage.Filter> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Filter stage) { Objects.requireNonNull(stage); Predicate predicate = Objects.requireNonNull(stage.getPredicate()); return source -> (Multi<O>) source.transform().byFilteringItemsWith(predicate); } }
FilterStageFactory implements ProcessingStageFactory<Stage.Filter> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Filter stage) { Objects.requireNonNull(stage); Predicate predicate = Objects.requireNonNull(stage.getPredicate()); return source -> (Multi<O>) source.transform().byFilteringItemsWith(predicate); }  }
FilterStageFactory implements ProcessingStageFactory<Stage.Filter> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Filter stage) { Objects.requireNonNull(stage); Predicate predicate = Objects.requireNonNull(stage.getPredicate()); return source -> (Multi<O>) source.transform().byFilteringItemsWith(predicate); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Filter stage); }
FilterStageFactory implements ProcessingStageFactory<Stage.Filter> { @SuppressWarnings("unchecked") @Override public <I, O> ProcessingStage<I, O> create(Engine engine, Stage.Filter stage) { Objects.requireNonNull(stage); Predicate predicate = Objects.requireNonNull(stage.getPredicate()); return source -> (Multi<O>) source.transform().byFilteringItemsWith(predicate); }  @SuppressWarnings("unchecked") @Override ProcessingStage<I, O> create(Engine engine, Stage.Filter stage);  }