TopologyDag implements Serializable { public List<Edge> getAllEdges() { List<Edge> result = new ArrayList<>(); for (List<Edge> edges : dag.values()) { result.addAll(edges); } return result; }  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 testGetAllEdges() throws Exception { Source src = new StreamlineSource(Collections.singleton(new Stream("f1", "f2"))); Sink sink = new StreamlineSink(); Processor processor = new StreamlineProcessor(Collections.singleton(new Stream("r1"))); topology.add(src).add(processor).add(sink); topology.addEdge(src, processor); topology.addEdge(processor, sink); assertEquals(Sets.newHashSet(src, processor), topology.getOutputComponents()); assertEquals(Sets.newHashSet(sink, processor), topology.getInputComponents()); assertEquals(2, topology.getAllEdges().size()); }