class Engine implements ReactiveStreamsEngine {

    @Override
    public <T> Publisher<T> buildPublisher(Graph graph) {
        Flowable<T> flowable = null;
        for (Stage stage : graph.getStages()) {
            Operator operator = Stages.lookup(stage);
            if (flowable == null) {
                if (operator instanceof PublisherOperator) {
                    flowable = createPublisher(stage, (PublisherOperator) operator);
                } else {
                    throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage);
                }
            } else {
                if (operator instanceof ProcessorOperator) {
                    flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator);
                } else {
                    throw new IllegalArgumentException("Expecting a processor stage, got a " + stage);
                }
            }
        }
        return flowable;
    }

}