HazelcastJetInterpreterUtils { public static String displayNetworkFromDAG(DAG dag){ GraphResult.Graph graph = new GraphResult.Graph(); graph.setDirected(true); Map<String, Integer> nodeIds = new HashMap<>(); List<Node> nodes = new ArrayList<>(); AtomicInteger nodeCount = new AtomicInteger(1); dag.forEach(v -> { nodeIds.put(v.getName(), nodeCount.getAndIncrement()); Node node = new Node(); node.setId(nodeIds.get(v.getName())); if (v.getName().toLowerCase().contains("sink")) node.setLabel("Sink"); else if (v.getName().toLowerCase().contains("source")) node.setLabel("Source"); else node.setLabel("Transform"); Map<String, Object> data = new HashMap<>(); data.put("description", v.getName()); node.setData(data); nodes.add(node); }); graph.setNodes(nodes); Map<String, String> labels = new HashMap<>(); labels.put("Source", "#00317c"); labels.put("Transform", "#ff7600"); labels.put("Sink", "#00317c"); graph.setLabels(labels); Map<String, Integer> edgeIds = new HashMap<>(); List<Relationship> rels = new ArrayList<>(); AtomicInteger relCount = new AtomicInteger(1); dag.forEach(v -> { dag.getInboundEdges(v.getName()).forEach(e -> { String edgeName = e.getSourceName() + " to " + e.getDestName(); if (edgeIds.get(edgeName) == null) { edgeIds.put(edgeName, relCount.getAndIncrement()); Relationship rel = new Relationship(); rel.setId(edgeIds.get(edgeName)); rel.setSource(nodeIds.get(e.getSourceName())); rel.setTarget(nodeIds.get(e.getDestName())); Map<String, Object> data = new HashMap<>(); data.put("routing", e.getRoutingPolicy().toString()); data.put("priority", e.getPriority()); data.put("distributed", e.isDistributed()); rel.setData(data); rels.add(rel); } }); dag.getOutboundEdges(v.getName()).forEach(e -> { String edgeName = e.getSourceName() + " to " + e.getDestName(); if (edgeIds.get(edgeName) == null) { edgeIds.put(edgeName, relCount.getAndIncrement()); Relationship rel = new Relationship(); rel.setId(edgeIds.get(edgeName)); rel.setSource(nodeIds.get(e.getSourceName())); rel.setTarget(nodeIds.get(e.getDestName())); Map<String, Object> data = new HashMap<>(); data.put("routing", e.getRoutingPolicy().toString()); data.put("priority", e.getPriority()); data.put("distributed", e.isDistributed()); rel.setData(data); rels.add(rel); } }); }); graph.setEdges(rels); return "%network " + gson.toJson(graph); }  static String displayNetworkFromDAG(DAG dag);  }
@Test public void testDisplayNetworkFromDAGUtil() { Pipeline p = Pipeline.create(); p.drawFrom(Sources.<String>list("text")) .flatMap(word -> traverseArray(word.toLowerCase().split("\\W+"))).setName("flat traversing") .filter(word -> !word.isEmpty()) .groupingKey(wholeItem()) .aggregate(counting()) .drainTo(Sinks.map("counts")); assertEquals( NETWORK_RESULT_1, HazelcastJetInterpreterUtils.displayNetworkFromDAG(p.toDag()) ); }