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; }  TopologyDag add(OutputComponent component); Set<OutputComponent> getOutputComponents(); TopologyDag add(InputComponent component); Set<InputComponent> getInputComponents(); TopologyDag add(Processor processor); Set<Component> getComponents(); void addEdge(OutputComponent from, InputComponent to); void addEdge(String id, OutputComponent from, InputComponent to, String streamId); void addEdge(String id, OutputComponent from, InputComponent to, String streamId, Stream.Grouping grouping); void addEdge(Edge edge); void addEdge(String id, OutputComponent from, InputComponent to, StreamGrouping streamGrouping); void removeEdge(OutputComponent from, InputComponent to); void removeEdge(OutputComponent from, InputComponent to, String streamId, Stream.Grouping grouping); List<Edge> getEdgesFrom(Component component); List<Edge> getEdgesTo(Component component); List<Edge> getEdges(Component component); List<Edge> getAllEdges(); List<Component> adjacent(Component component); void traverse(TopologyDagVisitor visitor); @Override String toString();  }
@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()); }