@Test(expected = IllegalArgumentException.class) public void testInvalidCompletion() { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.PublisherStage) Flowable::empty); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; engine.buildCompletion(graph); }
@Override public <T> CompletionStage<T> buildCompletion(Graph graph) { Flowable<?> 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) { Flowable<?> 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) { Flowable<?> 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) { Flowable<?> 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) { Flowable<?> 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);  }