class TopologyDag implements Serializable {

    List<Component> topOrder() {
        Map<Component, State> state = new HashMap<>();
        List<Component> res = new ArrayList<>();
        for(Component component: Sets.union(inputComponents, outputComponents)) {
            if (state.get(component) != State.VISITED) {
                res.addAll(dfs(component, state));
            }
        }
        Collections.reverse(res);
        return res;
    }

}

class TopologyDagTest {

     TopologyDag topology;

    @Test
    public void testTopOrder() {

        Source src1 = new StreamlineSource(Collections.singleton(new Stream("f1", "f2")));
        Source src2 = new StreamlineSource(Collections.singleton(new Stream("f1", "f2")));
        Sink sink = new StreamlineSink();
        Processor processor = new StreamlineProcessor(Collections.singleton(new Stream("r1")));

        topology.add(src1).add(src2).add(processor).add(sink);
        topology.addEdge(src1, processor);
        topology.addEdge(src2, processor);
        topology.addEdge(processor, sink);

        List<Component> topOrder = topology.topOrder();
        assertArrayEquals(new Component[] {src2, src1, processor, sink}, topOrder.toArray());
    }

}