@Test(expected = IllegalArgumentException.class) public void testInvalidSubscriber() { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; engine.buildSubscriber(graph); }
@Override public <T, R> SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph) { Processor<T, T> processor = new ConnectableProcessor<>(); Flowable<T> flowable = Flowable.fromPublisher(processor); for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else if (operator instanceof TerminalOperator) { CompletionStage<R> result = applySubscriber(Transformer.apply(flowable), stage, (TerminalOperator) operator); return new DefaultSubscriberWithCompletionStage<>(processor, result); } else { throw new UnsupportedStageException(stage); } } throw new IllegalArgumentException("The graph does not have a valid final stage"); }
Engine implements ReactiveStreamsEngine { @Override public <T, R> SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph) { Processor<T, T> processor = new ConnectableProcessor<>(); Flowable<T> flowable = Flowable.fromPublisher(processor); for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else if (operator instanceof TerminalOperator) { CompletionStage<R> result = applySubscriber(Transformer.apply(flowable), stage, (TerminalOperator) operator); return new DefaultSubscriberWithCompletionStage<>(processor, result); } else { throw new UnsupportedStageException(stage); } } throw new IllegalArgumentException("The graph does not have a valid final stage"); } }
Engine implements ReactiveStreamsEngine { @Override public <T, R> SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph) { Processor<T, T> processor = new ConnectableProcessor<>(); Flowable<T> flowable = Flowable.fromPublisher(processor); for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else if (operator instanceof TerminalOperator) { CompletionStage<R> result = applySubscriber(Transformer.apply(flowable), stage, (TerminalOperator) operator); return new DefaultSubscriberWithCompletionStage<>(processor, result); } else { throw new UnsupportedStageException(stage); } } throw new IllegalArgumentException("The graph does not have a valid final stage"); }  }
Engine implements ReactiveStreamsEngine { @Override public <T, R> SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph) { Processor<T, T> processor = new ConnectableProcessor<>(); Flowable<T> flowable = Flowable.fromPublisher(processor); for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else if (operator instanceof TerminalOperator) { CompletionStage<R> result = applySubscriber(Transformer.apply(flowable), stage, (TerminalOperator) operator); return new DefaultSubscriberWithCompletionStage<>(processor, result); } else { throw new UnsupportedStageException(stage); } } throw new IllegalArgumentException("The graph does not have a valid final 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, R> SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph) { Processor<T, T> processor = new ConnectableProcessor<>(); Flowable<T> flowable = Flowable.fromPublisher(processor); for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else if (operator instanceof TerminalOperator) { CompletionStage<R> result = applySubscriber(Transformer.apply(flowable), stage, (TerminalOperator) operator); return new DefaultSubscriberWithCompletionStage<>(processor, result); } else { throw new UnsupportedStageException(stage); } } throw new IllegalArgumentException("The graph does not have a valid final 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);  }