@Test (dependsOnMethods = "testUnresolvedFlow") public void testGitFlowGraphMonitorService() throws IOException, GitAPIException, URISyntaxException, InterruptedException { File remoteDir = new File(TESTDIR + "/remote"); File cloneDir = new File(TESTDIR + "/clone"); File flowGraphDir = new File(cloneDir, "/gobblin-flowgraph"); cleanUpDir(TESTDIR); RepositoryCache.FileKey fileKey = RepositoryCache.FileKey.exact(remoteDir, FS.DETECTED); Repository remoteRepo = fileKey.open(false); remoteRepo.create(true); Git gitForPush = Git.cloneRepository().setURI(remoteRepo.getDirectory().getAbsolutePath()).setDirectory(cloneDir).call(); gitForPush.commit().setMessage("First commit").call(); RefSpec masterRefSpec = new RefSpec("master"); gitForPush.push().setRemote("origin").setRefSpecs(masterRefSpec).call(); URI flowTemplateCatalogUri = this.getClass().getClassLoader().getResource("template_catalog").toURI(); Config config = ConfigBuilder.create() .addPrimitive(GitFlowGraphMonitor.GIT_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.GIT_MONITOR_REPO_URI, remoteRepo.getDirectory().getAbsolutePath()) .addPrimitive(GitFlowGraphMonitor.GIT_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.GIT_MONITOR_REPO_DIR, TESTDIR + "/git-flowgraph") .addPrimitive(GitFlowGraphMonitor.GIT_FLOWGRAPH_MONITOR_PREFIX + "." + ConfigurationKeys.GIT_MONITOR_POLLING_INTERVAL, 5) .addPrimitive(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY, flowTemplateCatalogUri.toString()) .build(); specCompiler = new MultiHopFlowCompiler(config, Optional.absent(), false); specCompiler.setActive(true); Assert.assertNull(specCompiler.getFlowGraph().getNode("node1")); File nodeDir = new File(flowGraphDir, "node1"); File nodeFile = new File(nodeDir, "node1.properties"); nodeDir.mkdirs(); nodeFile.createNewFile(); Files.write(FlowGraphConfigurationKeys.DATA_NODE_IS_ACTIVE_KEY + "=true\nparam1=val1" + "\n", nodeFile, Charsets.UTF_8); gitForPush.add().addFilepattern(formNodeFilePath(flowGraphDir, nodeDir.getName(), nodeFile.getName())).call(); gitForPush.commit().setMessage("Node commit").call(); gitForPush.push().setRemote("origin").setRefSpecs(masterRefSpec).call(); TimeUnit.SECONDS.sleep(10); DataNode dataNode = specCompiler.getFlowGraph().getNode("node1"); Assert.assertEquals(dataNode.getId(), "node1"); Assert.assertEquals(dataNode.getRawConfig().getString("param1"), "val1"); }
@Override public void setActive(boolean active) { super.setActive(active); if (this.gitFlowGraphMonitor != null) { this.gitFlowGraphMonitor.setActive(active); } }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public void setActive(boolean active) { super.setActive(active); if (this.gitFlowGraphMonitor != null) { this.gitFlowGraphMonitor.setActive(active); } } }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public void setActive(boolean active) { super.setActive(active); if (this.gitFlowGraphMonitor != null) { this.gitFlowGraphMonitor.setActive(active); } } MultiHopFlowCompiler(Config config);  MultiHopFlowCompiler(Config config, boolean instrumentationEnabled);  MultiHopFlowCompiler(Config config, Optional<Logger> log);  MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled); @VisibleForTesting  MultiHopFlowCompiler(Config config, FlowGraph flowGraph); }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public void setActive(boolean active) { super.setActive(active); if (this.gitFlowGraphMonitor != null) { this.gitFlowGraphMonitor.setActive(active); } } MultiHopFlowCompiler(Config config);  MultiHopFlowCompiler(Config config, boolean instrumentationEnabled);  MultiHopFlowCompiler(Config config, Optional<Logger> log);  MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled); @VisibleForTesting  MultiHopFlowCompiler(Config config, FlowGraph flowGraph); @Override void setActive(boolean active); @Override void awaitHealthy(); @Override Dag<JobExecutionPlan> compileFlow(Spec spec); }
MultiHopFlowCompiler extends BaseFlowToJobSpecCompiler { @Override public void setActive(boolean active) { super.setActive(active); if (this.gitFlowGraphMonitor != null) { this.gitFlowGraphMonitor.setActive(active); } } MultiHopFlowCompiler(Config config);  MultiHopFlowCompiler(Config config, boolean instrumentationEnabled);  MultiHopFlowCompiler(Config config, Optional<Logger> log);  MultiHopFlowCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled); @VisibleForTesting  MultiHopFlowCompiler(Config config, FlowGraph flowGraph); @Override void setActive(boolean active); @Override void awaitHealthy(); @Override Dag<JobExecutionPlan> compileFlow(Spec spec);  }