@Test (dependsOnMethods = "testRemoveNode") public void testChangesReorder() throws GitAPIException, IOException, ExecutionException, InterruptedException { String node1FileContents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam1=value1\n"; String node2FileContents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam2=value2\n"; String edgeFileContents = buildEdgeFileContents("node1", "node2", "edge1", "value1"); createNewFile(this.node1Dir, this.node1File, node1FileContents); createNewFile(this.node2Dir, this.node2File, node2FileContents); createNewFile(this.edge1Dir, this.edge1File, edgeFileContents); this.gitForPush.add().addFilepattern(formNodeFilePath(this.node1Dir.getName(), this.node1File.getName())).call(); this.gitForPush.add().addFilepattern(formNodeFilePath(this.node2Dir.getName(), this.node2File.getName())).call(); this.gitForPush.commit().setMessage("Add nodes commit").call(); this.gitForPush.push().setRemote("origin").setRefSpecs(this.masterRefSpec).call(); this.gitForPush.add().addFilepattern(formEdgeFilePath(this.edge1Dir.getParentFile().getName(), this.edge1Dir.getName(), this.edge1File.getName())).call(); this.gitForPush.commit().setMessage("Add nodes and edges commit").call(); this.gitForPush.push().setRemote("origin").setRefSpecs(this.masterRefSpec).call(); this.gitFlowGraphMonitor.processGitConfigChanges(); DataNode node1 = this.flowGraph.getNode("node1"); Assert.assertNotNull(node1); DataNode node2 = this.flowGraph.getNode("node2"); Assert.assertNotNull(node2); testIfEdgeSuccessfullyAdded("node1", "node2", "edge1", "value1"); node1File.delete(); edge1File.delete(); this.gitForPush.rm().addFilepattern(formNodeFilePath(this.node1Dir.getName(), this.node1File.getName())).call(); this.gitForPush.rm().addFilepattern(formEdgeFilePath(this.edge1Dir.getParentFile().getName(), this.edge1Dir.getName(), this.edge1File.getName())).call(); this.gitForPush.commit().setMessage("Delete node1 and edge1 commit").call(); this.gitForPush.push().setRemote("origin").setRefSpecs(this.masterRefSpec).call(); createNewFile(this.node1Dir, this.node1File, node1FileContents); this.gitForPush.add().addFilepattern(formNodeFilePath(this.node1Dir.getName(), this.node1File.getName())).call(); this.gitForPush.commit().setMessage("Add node1 commit").call(); this.gitForPush.push().setRemote("origin").setRefSpecs(this.masterRefSpec).call(); this.gitFlowGraphMonitor.processGitConfigChanges(); node1 = this.flowGraph.getNode("node1"); Assert.assertNotNull(node1); Assert.assertEquals(this.flowGraph.getEdges(node1).size(), 0); }
@Override void processGitConfigChanges() throws GitAPIException, IOException { if (flowTemplateCatalog.isPresent() && flowTemplateCatalog.get().getAndSetShouldRefreshFlowGraph(false)) { log.info("Change to template catalog detected, refreshing FlowGraph"); this.gitRepo.initRepository(); } List<DiffEntry> changes = this.gitRepo.getChanges(); Collections.sort(changes, (o1, o2) -> { Integer o1Depth = (o1.getNewPath() != null) ? (new Path(o1.getNewPath())).depth() : (new Path(o1.getOldPath())).depth(); Integer o2Depth = (o2.getNewPath() != null) ? (new Path(o2.getNewPath())).depth() : (new Path(o2.getOldPath())).depth(); return o1Depth.compareTo(o2Depth); }); processGitConfigChangesHelper(changes); this.initComplete.countDown(); }
GitFlowGraphMonitor extends GitMonitoringService { @Override void processGitConfigChanges() throws GitAPIException, IOException { if (flowTemplateCatalog.isPresent() && flowTemplateCatalog.get().getAndSetShouldRefreshFlowGraph(false)) { log.info("Change to template catalog detected, refreshing FlowGraph"); this.gitRepo.initRepository(); } List<DiffEntry> changes = this.gitRepo.getChanges(); Collections.sort(changes, (o1, o2) -> { Integer o1Depth = (o1.getNewPath() != null) ? (new Path(o1.getNewPath())).depth() : (new Path(o1.getOldPath())).depth(); Integer o2Depth = (o2.getNewPath() != null) ? (new Path(o2.getNewPath())).depth() : (new Path(o2.getOldPath())).depth(); return o1Depth.compareTo(o2Depth); }); processGitConfigChangesHelper(changes); this.initComplete.countDown(); } }
GitFlowGraphMonitor extends GitMonitoringService { @Override void processGitConfigChanges() throws GitAPIException, IOException { if (flowTemplateCatalog.isPresent() && flowTemplateCatalog.get().getAndSetShouldRefreshFlowGraph(false)) { log.info("Change to template catalog detected, refreshing FlowGraph"); this.gitRepo.initRepository(); } List<DiffEntry> changes = this.gitRepo.getChanges(); Collections.sort(changes, (o1, o2) -> { Integer o1Depth = (o1.getNewPath() != null) ? (new Path(o1.getNewPath())).depth() : (new Path(o1.getOldPath())).depth(); Integer o2Depth = (o2.getNewPath() != null) ? (new Path(o2.getNewPath())).depth() : (new Path(o2.getOldPath())).depth(); return o1Depth.compareTo(o2Depth); }); processGitConfigChangesHelper(changes); this.initComplete.countDown(); } GitFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
      FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete); }
GitFlowGraphMonitor extends GitMonitoringService { @Override void processGitConfigChanges() throws GitAPIException, IOException { if (flowTemplateCatalog.isPresent() && flowTemplateCatalog.get().getAndSetShouldRefreshFlowGraph(false)) { log.info("Change to template catalog detected, refreshing FlowGraph"); this.gitRepo.initRepository(); } List<DiffEntry> changes = this.gitRepo.getChanges(); Collections.sort(changes, (o1, o2) -> { Integer o1Depth = (o1.getNewPath() != null) ? (new Path(o1.getNewPath())).depth() : (new Path(o1.getOldPath())).depth(); Integer o2Depth = (o2.getNewPath() != null) ? (new Path(o2.getNewPath())).depth() : (new Path(o2.getOldPath())).depth(); return o1Depth.compareTo(o2Depth); }); processGitConfigChangesHelper(changes); this.initComplete.countDown(); } GitFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
      FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete); @Override boolean shouldPollGit(); @Override void addChange(DiffEntry change); @Override void removeChange(DiffEntry change); }
GitFlowGraphMonitor extends GitMonitoringService { @Override void processGitConfigChanges() throws GitAPIException, IOException { if (flowTemplateCatalog.isPresent() && flowTemplateCatalog.get().getAndSetShouldRefreshFlowGraph(false)) { log.info("Change to template catalog detected, refreshing FlowGraph"); this.gitRepo.initRepository(); } List<DiffEntry> changes = this.gitRepo.getChanges(); Collections.sort(changes, (o1, o2) -> { Integer o1Depth = (o1.getNewPath() != null) ? (new Path(o1.getNewPath())).depth() : (new Path(o1.getOldPath())).depth(); Integer o2Depth = (o2.getNewPath() != null) ? (new Path(o2.getNewPath())).depth() : (new Path(o2.getOldPath())).depth(); return o1Depth.compareTo(o2Depth); }); processGitConfigChangesHelper(changes); this.initComplete.countDown(); } GitFlowGraphMonitor(Config config, Optional<? extends FSFlowTemplateCatalog> flowTemplateCatalog,
      FlowGraph graph, Map<URI, TopologySpec> topologySpecMap, CountDownLatch initComplete); @Override boolean shouldPollGit(); @Override void addChange(DiffEntry change); @Override void removeChange(DiffEntry change); static final String GIT_FLOWGRAPH_MONITOR_PREFIX; }