class 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");
    }

}