LogicalDagOptimizer { public DAG<MISTStream, MISTEdge> getOptimizedDAG() { final Set<MISTStream> visited = new HashSet<>(); for (final MISTStream source : dag.getRootVertices()) { final Map<MISTStream, MISTEdge> rootEdges = dag.getEdges(source); visited.add(source); for (final MISTStream nextVertex : rootEdges.keySet()) { optimizeSubDag(nextVertex, visited); } } return dag; } LogicalDagOptimizer(final DAG<MISTStream, MISTEdge> dag); DAG<MISTStream, MISTEdge> getOptimizedDAG();  }
@Test public void testConditionalBranch() throws InjectionException { final MISTQueryBuilder queryBuilder = new MISTQueryBuilder(); queryBuilder.setApplicationId(TestParameters.SUPER_GROUP_ID); final ContinuousStream<String> src1 = queryBuilder.socketTextStream(TestParameters.LOCAL_TEXT_SOCKET_SOURCE_CONF); final ContinuousStream<String> op1 = src1.filter((x) -> true); final ContinuousStream<String> cbr1 = op1.routeIf((x) -> true); final ContinuousStream<String> cbr2 = op1.routeIf((x) -> true); final ContinuousStream<String> cbr3 = op1.routeIf((x) -> true); final ContinuousStream<String> op2 = cbr1.filter((x) -> true); final ContinuousStream<String> op3 = cbr1.filter((x) -> true); final ContinuousStream<String> op4 = cbr3.filter((x) -> true); final ContinuousStream<String> union = op4.union(cbr2); final MISTStream<String> sink1 = op2.textSocketOutput(TestParameters.HOST, TestParameters.SINK_PORT); final MISTStream<String> sink2 = op3.textSocketOutput(TestParameters.HOST, TestParameters.SINK_PORT); final MISTStream<String> sink3 = union.textSocketOutput(TestParameters.HOST, TestParameters.SINK_PORT); final MISTQuery query = queryBuilder.build(); final DAG<MISTStream, MISTEdge> dag = query.getDAG(); final LogicalDagOptimizer logicalDagOptimizer = new LogicalDagOptimizer(dag); final DAG<MISTStream, MISTEdge> optimizedDAG = logicalDagOptimizer.getOptimizedDAG(); final Map<MISTStream, MISTEdge> e1 = optimizedDAG.getEdges(src1); final Map<MISTStream, MISTEdge> result1 = new HashMap<>(); result1.put(op1, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e1, result1); final Map<MISTStream, MISTEdge> e2 = optimizedDAG.getEdges(op1); Assert.assertEquals(1, e2.size()); final MISTStream cbrOp = e2.entrySet().iterator().next().getKey(); Assert.assertTrue(cbrOp instanceof ContinuousStreamImpl); final Map<MISTStream, MISTEdge> e3 = optimizedDAG.getEdges(cbrOp); final Map<MISTStream, MISTEdge> result3 = new HashMap<>(); result3.put(op2, new MISTEdge(Direction.LEFT, 1)); result3.put(op3, new MISTEdge(Direction.LEFT, 1)); result3.put(union, new MISTEdge(Direction.RIGHT, 2)); result3.put(op4, new MISTEdge(Direction.LEFT, 3)); Assert.assertEquals(e3, result3); final Map<MISTStream, MISTEdge> e4 = optimizedDAG.getEdges(op4); final Map<MISTStream, MISTEdge> result4 = new HashMap<>(); result4.put(union, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e4, result4); final Map<MISTStream, MISTEdge> e5 = optimizedDAG.getEdges(op2); final Map<MISTStream, MISTEdge> result5 = new HashMap<>(); result5.put(sink1, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e5, result5); final Map<MISTStream, MISTEdge> e6 = optimizedDAG.getEdges(op3); final Map<MISTStream, MISTEdge> result6 = new HashMap<>(); result6.put(sink2, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e6, result6); final Map<MISTStream, MISTEdge> e7 = optimizedDAG.getEdges(union); final Map<MISTStream, MISTEdge> result7 = new HashMap<>(); result7.put(sink3, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e7, result7); }