DeduceFlattenLocationsFunction implements Function<MutableNetwork<Node, Edge>, MutableNetwork<Node, Edge>> { @Override public MutableNetwork<Node, Edge> apply(MutableNetwork<Node, Edge> network) { Map<Node, AggregatedLocation> predecessorLocationsMap = new HashMap<>(); Map<Node, AggregatedLocation> successorLocationsMap = new HashMap<>(); Map<Node, ExecutionLocation> deducedLocationsMap = new HashMap<>(); ImmutableList<Node> flattens = ImmutableList.copyOf(Iterables.filter(network.nodes(), IsFlatten.INSTANCE)); for (Node flatten : flattens) { AggregatedLocation predecessorLocations = AggregatedLocation.NEITHER; AggregatedLocation successorLocations = AggregatedLocation.NEITHER; predecessorLocations = getPredecessorLocations(flatten, network, predecessorLocationsMap); successorLocations = getSuccessorLocations(flatten, network, successorLocationsMap); deducedLocationsMap.put( flatten, DEDUCTION_TABLE.get(predecessorLocations, successorLocations)); } Networks.replaceDirectedNetworkNodes( network, (Node node) -> { if (!deducedLocationsMap.containsKey(node)) { return node; } ParallelInstructionNode castNode = ((ParallelInstructionNode) node); ExecutionLocation deducedLocation = deducedLocationsMap.get(node); return ParallelInstructionNode.create(castNode.getParallelInstruction(), deducedLocation); }); return network; }  @Override MutableNetwork<Node, Edge> apply(MutableNetwork<Node, Edge> network);  }
@Test public void testEmptyNetwork() throws Exception { assertEquals( createEmptyNetwork(), new DeduceFlattenLocationsFunction().apply(createEmptyNetwork())); }
@Test public void testDeductionOfChainedFlattens() throws Exception { MutableNetwork<Node, Edge> network = createEmptyNetwork(); Node sdkNode1 = createSdkNode("sdk_node1"); Node sdkNode1Output = createPCollection("sdk_node1.out"); Node sdkNode2 = createSdkNode("sdk_node2"); Node sdkNode2Output = createPCollection("sdk_node2.out"); Node sdkNode3 = createSdkNode("sdk_node3"); Node sdkNode3Output = createPCollection("sdk_node3.out"); Node runnerNode1 = createRunnerNode("runner_node1"); Node runnerNode1Output = createPCollection("runner_node1.out"); Node runnerNode2 = createRunnerNode("runner_node2"); Node runnerNode2Output = createPCollection("runner_node2.out"); Node runnerNode3 = createRunnerNode("runner_node3"); Node runnerNode3Output = createPCollection("runner_node3.out"); Node flatten1 = createFlatten("flatten1"); Node flatten1Output = createPCollection("flatten1.out"); Node flatten2 = createFlatten("flatten2"); Node flatten2Output = createPCollection("flatten2.out"); Node flatten3 = createFlatten("flatten3"); Node flatten3Output = createPCollection("flatten3.out"); network.addNode(sdkNode1); network.addNode(sdkNode2); network.addNode(sdkNode3); network.addNode(runnerNode1); network.addNode(runnerNode2); network.addNode(runnerNode3); network.addNode(flatten1); network.addNode(flatten1Output); network.addNode(flatten2); network.addNode(flatten2Output); network.addNode(flatten3); network.addNode(flatten3Output); network.addEdge(sdkNode1, sdkNode1Output, DefaultEdge.create()); network.addEdge(sdkNode2, sdkNode2Output, DefaultEdge.create()); network.addEdge(runnerNode1, runnerNode1Output, DefaultEdge.create()); network.addEdge(runnerNode2, runnerNode2Output, DefaultEdge.create()); network.addEdge(sdkNode1Output, flatten1, DefaultEdge.create()); network.addEdge(sdkNode2Output, flatten1, DefaultEdge.create()); network.addEdge(runnerNode1Output, flatten2, DefaultEdge.create()); network.addEdge(runnerNode2Output, flatten2, DefaultEdge.create()); network.addEdge(flatten1, flatten1Output, DefaultEdge.create()); network.addEdge(flatten2, flatten2Output, DefaultEdge.create()); network.addEdge(flatten1Output, flatten3, DefaultEdge.create()); network.addEdge(flatten2Output, flatten3, DefaultEdge.create()); network.addEdge(flatten3, flatten3Output, DefaultEdge.create()); network.addEdge(flatten3Output, sdkNode3, DefaultEdge.create()); network.addEdge(flatten3Output, runnerNode3, DefaultEdge.create()); network.addEdge(sdkNode3, sdkNode3Output, DefaultEdge.create()); network.addEdge(runnerNode3, runnerNode3Output, DefaultEdge.create()); network = new DeduceFlattenLocationsFunction().apply(network); ExecutionLocation flatten1Location = getExecutionLocationOf("flatten1", network); assertEquals(flatten1Location, ExecutionLocation.SDK_HARNESS); ExecutionLocation flatten2Location = getExecutionLocationOf("flatten2", network); assertEquals(flatten2Location, ExecutionLocation.RUNNER_HARNESS); ExecutionLocation flatten3Location = getExecutionLocationOf("flatten3", network); assertEquals(flatten3Location, ExecutionLocation.AMBIGUOUS); }