@Test public void testCreatingPublisherWithUnknownStage() { assertThrows(UnsupportedStageException.class, () -> { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add(new Stage() { }); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; engine.buildPublisher(graph); }); }
@Override public <T> Publisher<T> buildPublisher(Graph graph) { Multi<T> publisher = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (publisher == null) { if (operator instanceof PublisherOperator) { publisher = createPublisher(stage, (PublisherOperator) operator); } else { throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage); } } else { if (operator instanceof ProcessorOperator) { publisher = applyProcessors(publisher, stage, (ProcessorOperator) operator); } else { throw new IllegalArgumentException("Expecting a processor stage, got a " + stage); } } } return publisher; }
Engine implements ReactiveStreamsEngine { @Override public <T> Publisher<T> buildPublisher(Graph graph) { Multi<T> publisher = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (publisher == null) { if (operator instanceof PublisherOperator) { publisher = createPublisher(stage, (PublisherOperator) operator); } else { throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage); } } else { if (operator instanceof ProcessorOperator) { publisher = applyProcessors(publisher, stage, (ProcessorOperator) operator); } else { throw new IllegalArgumentException("Expecting a processor stage, got a " + stage); } } } return publisher; } }
Engine implements ReactiveStreamsEngine { @Override public <T> Publisher<T> buildPublisher(Graph graph) { Multi<T> publisher = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (publisher == null) { if (operator instanceof PublisherOperator) { publisher = createPublisher(stage, (PublisherOperator) operator); } else { throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage); } } else { if (operator instanceof ProcessorOperator) { publisher = applyProcessors(publisher, stage, (ProcessorOperator) operator); } else { throw new IllegalArgumentException("Expecting a processor stage, got a " + stage); } } } return publisher; }  }
Engine implements ReactiveStreamsEngine { @Override public <T> Publisher<T> buildPublisher(Graph graph) { Multi<T> publisher = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (publisher == null) { if (operator instanceof PublisherOperator) { publisher = createPublisher(stage, (PublisherOperator) operator); } else { throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage); } } else { if (operator instanceof ProcessorOperator) { publisher = applyProcessors(publisher, stage, (ProcessorOperator) operator); } else { throw new IllegalArgumentException("Expecting a processor stage, got a " + stage); } } } return publisher; }  @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> Publisher<T> buildPublisher(Graph graph) { Multi<T> publisher = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (publisher == null) { if (operator instanceof PublisherOperator) { publisher = createPublisher(stage, (PublisherOperator) operator); } else { throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage); } } else { if (operator instanceof ProcessorOperator) { publisher = applyProcessors(publisher, stage, (ProcessorOperator) operator); } else { throw new IllegalArgumentException("Expecting a processor stage, got a " + stage); } } } return publisher; }  @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);  }