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

}