PipelineDotRenderer implements Pipeline.PipelineVisitor { public static String toDotString(Pipeline pipeline) { final PipelineDotRenderer visitor = new PipelineDotRenderer(); visitor.begin(); pipeline.traverseTopologically(visitor); visitor.end(); return visitor.dotBuilder.toString(); } private  PipelineDotRenderer(); static String toDotString(Pipeline pipeline); static String toDotString(RunnerApi.Pipeline pipeline); @Override void enterPipeline(Pipeline p); @Override void leavePipeline(Pipeline pipeline); @Override CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node); @Override void leaveCompositeTransform(TransformHierarchy.Node node); @Override void visitPrimitiveTransform(TransformHierarchy.Node node); @Override void visitValue(PValue value, TransformHierarchy.Node producer);  }
@Test public void testEmptyPipeline() { assertEquals( "digraph {" + " rankdir=LR" + " subgraph cluster_0 {" + " label = \"\"" + " }" + "}", PipelineDotRenderer.toDotString(p).replaceAll(System.lineSeparator(), "")); }
@Test public void testCompositePipeline() { p.apply(Create.timestamped(TimestampedValue.of(KV.of(1, 1), new Instant(1)))) .apply(Window.into(FixedWindows.of(Duration.millis(10)))) .apply(Sum.integersPerKey()); assertEquals( "digraph {" + " rankdir=LR" + " subgraph cluster_0 {" + " label = \"\"" + " subgraph cluster_1 {" + " label = \"Create.TimestampedValues\"" + " subgraph cluster_2 {" + " label = \"Create.TimestampedValues/Create.Values\"" + " subgraph cluster_3 {" + " label = \"Create.TimestampedValues/Create.Values/Read(CreateSource)\"" + " 4 [label=\"Impulse\"]" + " subgraph cluster_5 {" + " label = \"Create.TimestampedValues/Create.Values/Read(CreateSource)/ParDo(OutputSingleSource)\"" + " 6 [label=\"ParMultiDo(OutputSingleSource)\"]" + " 4 -> 6 [style=solid label=\"\"]" + " }" + " subgraph cluster_7 {" + " label = \"Create.TimestampedValues/Create.Values/Read(CreateSource)/ParDo(BoundedSourceAsSDFWrapper)\"" + " 8 [label=\"ParMultiDo(BoundedSourceAsSDFWrapper)\"]" + " 6 -> 8 [style=solid label=\"\"]" + " }" + " }" + " }" + " subgraph cluster_9 {" + " label = \"Create.TimestampedValues/ParDo(ConvertTimestamps)\"" + " 10 [label=\"ParMultiDo(ConvertTimestamps)\"]" + " 8 -> 10 [style=solid label=\"\"]" + " }" + " }" + " subgraph cluster_11 {" + " label = \"Window.Into()\"" + " 12 [label=\"Window.Assign\"]" + " 10 -> 12 [style=solid label=\"\"]" + " }" + " subgraph cluster_13 {" + " label = \"Combine.perKey(SumInteger)\"" + " 14 [label=\"GroupByKey\"]" + " 12 -> 14 [style=solid label=\"\"]" + " subgraph cluster_15 {" + " label = \"Combine.perKey(SumInteger)/Combine.GroupedValues\"" + " subgraph cluster_16 {" + " label = \"Combine.perKey(SumInteger)/Combine.GroupedValues/ParDo(Anonymous)\"" + " 17 [label=\"ParMultiDo(Anonymous)\"]" + " 14 -> 17 [style=solid label=\"\"]" + " }" + " }" + " }" + " }" + "}", PipelineDotRenderer.toDotString(p).replaceAll(System.lineSeparator(), "")); }
@Test public void testCompositePipeline() { p.apply(Create.timestamped(TimestampedValue.of(KV.of(1, 1), new Instant(1)))) .apply(Window.into(FixedWindows.of(Duration.millis(10)))) .apply(Sum.integersPerKey()); assertEquals( "digraph {" + " rankdir=LR" + " subgraph cluster_0 {" + " label = \"\"" + " subgraph cluster_1 {" + " label = \"Create.TimestampedValues\"" + " subgraph cluster_2 {" + " label = \"Create.TimestampedValues/Create.Values\"" + " 3 [label=\"Read(CreateSource)\"]" + " }" + " subgraph cluster_4 {" + " label = \"Create.TimestampedValues/ParDo(ConvertTimestamps)\"" + " 5 [label=\"ParMultiDo(ConvertTimestamps)\"]" + " 3 -> 5 [style=solid label=\"\"]" + " }" + " }" + " subgraph cluster_6 {" + " label = \"Window.Into()\"" + " 7 [label=\"Window.Assign\"]" + " 5 -> 7 [style=solid label=\"\"]" + " }" + " subgraph cluster_8 {" + " label = \"Combine.perKey(SumInteger)\"" + " 9 [label=\"GroupByKey\"]" + " 7 -> 9 [style=solid label=\"\"]" + " subgraph cluster_10 {" + " label = \"Combine.perKey(SumInteger)/Combine.GroupedValues\"" + " subgraph cluster_11 {" + " label = \"Combine.perKey(SumInteger)/Combine.GroupedValues/ParDo(Anonymous)\"" + " 12 [label=\"ParMultiDo(Anonymous)\"]" + " 9 -> 12 [style=solid label=\"\"]" + " }" + " }" + " }" + " }" + "}", PipelineDotRenderer.toDotString(p).replaceAll(System.lineSeparator(), "")); }