@Test (dependsOnMethods = "testUpdateEdge") public void testUpdateNode() throws IOException, GitAPIException, URISyntaxException, ExecutionException, InterruptedException { String fileContents = FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam1=value3\n"; addNode(this.node1Dir, this.node1File, fileContents); this.gitFlowGraphMonitor.processGitConfigChanges(); DataNode dataNode = this.flowGraph.getNode("node1"); Assert.assertEquals(dataNode.getId(), "node1"); Assert.assertTrue(dataNode.isActive()); Assert.assertEquals(dataNode.getRawConfig().getString("param1"), "value3"); }
@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; }