@Test public void testInvalidCompletion() { assertThrows(IllegalArgumentException.class, () -> { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.PublisherStage) () -> Multi.createFrom().empty()); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; engine.buildCompletion(graph); }); }
@Override public <T> CompletionStage<T> buildCompletion(Graph graph) { Multi<?> flowable = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof PublisherOperator) { flowable = createPublisher(stage, (PublisherOperator) operator); } else if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else { return applySubscriber(flowable, stage, (TerminalOperator) operator); } } throw new IllegalArgumentException("Graph did not have terminal stage"); }
Engine implements ReactiveStreamsEngine { @Override public <T> CompletionStage<T> buildCompletion(Graph graph) { Multi<?> flowable = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof PublisherOperator) { flowable = createPublisher(stage, (PublisherOperator) operator); } else if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else { return applySubscriber(flowable, stage, (TerminalOperator) operator); } } throw new IllegalArgumentException("Graph did not have terminal stage"); } }
Engine implements ReactiveStreamsEngine { @Override public <T> CompletionStage<T> buildCompletion(Graph graph) { Multi<?> flowable = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof PublisherOperator) { flowable = createPublisher(stage, (PublisherOperator) operator); } else if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else { return applySubscriber(flowable, stage, (TerminalOperator) operator); } } throw new IllegalArgumentException("Graph did not have terminal stage"); }  }
Engine implements ReactiveStreamsEngine { @Override public <T> CompletionStage<T> buildCompletion(Graph graph) { Multi<?> flowable = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof PublisherOperator) { flowable = createPublisher(stage, (PublisherOperator) operator); } else if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else { return applySubscriber(flowable, stage, (TerminalOperator) operator); } } throw new IllegalArgumentException("Graph did not have terminal stage"); }  @Override Publisher<T> buildPublisher(Graph graph); @Override SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph); @Override Processor<T, R> buildProcessor(Graph graph); @Override CompletionStage<T> buildCompletion(Graph graph); }
Engine implements ReactiveStreamsEngine { @Override public <T> CompletionStage<T> buildCompletion(Graph graph) { Multi<?> flowable = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof PublisherOperator) { flowable = createPublisher(stage, (PublisherOperator) operator); } else if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else { return applySubscriber(flowable, stage, (TerminalOperator) operator); } } throw new IllegalArgumentException("Graph did not have terminal stage"); }  @Override Publisher<T> buildPublisher(Graph graph); @Override SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph); @Override Processor<T, R> buildProcessor(Graph graph); @Override CompletionStage<T> buildCompletion(Graph graph);  }