LegacyFlowManagerService { public JobTicket runFlow(LegacyFlow flow, int batchSize, int threadCount, Map<String, Object> options, LegacyFlowStatusListener statusListener) { LegacyFlowRunner flowRunner = flowManager.newFlowRunner() .withFlow(flow) .withOptions(options) .withBatchSize(batchSize) .withThreadCount(threadCount) .onStatusChanged(statusListener); return flowRunner.run(); }  List<FlowModel> getFlows(String entityName, FlowType flowType); LegacyFlow getServerFlow(String entityName, String flowName, FlowType flowType); JobTicket runFlow(LegacyFlow flow, int batchSize, int threadCount, Map<String, Object> options, LegacyFlowStatusListener statusListener); Map<String, Object> getHarmonizeFlowOptionsFromFile(String entityName, String flowName); void saveOrUpdateHarmonizeFlowOptionsToFile(String entityName, String flowName, String harmonizeOptionsFileContent); void saveOrUpdateFlowMlcpOptionsToFile(String entityName, String flowName, String mlcpOptionsFileContent); Map<String, Object> getFlowMlcpOptionsFromFile(String entityName, String flowName); void runMlcp(LegacyFlow flow, JsonNode json, LegacyFlowStatusListener statusListener); LegacyFlowManager getFlowManager();  }
@Test public void runHarmonizationFlow() throws InterruptedException { clearDatabases(HubConfig.DEFAULT_STAGING_NAME, HubConfig.DEFAULT_FINAL_NAME, HubConfig.DEFAULT_JOB_NAME); assertEquals(0, getFinalDocCount()); DocumentMetadataHandle meta = new DocumentMetadataHandle(); meta.getCollections().add(ENTITY); installStagingDoc("/staged.json", meta, "legacy-flow-manager/staged.json"); String pdir = "C:\\some\\crazy\\path\\to\\project"; setEnvConfig(); String flowName = "sjs-json-harmonization-flow"; LegacyFlow flow = flowManager.getFlow(ENTITY, flowName, FlowType.HARMONIZE); Object monitor = new Object(); JobTicket jobTicket = fm.runFlow(flow, 1, 1, null, new LegacyFlowStatusListener(){ @Override public void onStatusChange(String jobId, int percentComplete, String message) { if (percentComplete == 100) { synchronized (monitor) { monitor.notify(); } } } }); synchronized (monitor) { monitor.wait(); } assertEquals(1, getFinalDocCount()); }
@Test public void runHarmonizationFlowWithOptions() throws InterruptedException { clearDatabases(HubConfig.DEFAULT_STAGING_NAME, HubConfig.DEFAULT_FINAL_NAME, HubConfig.DEFAULT_JOB_NAME); assertEquals(0, getFinalDocCount()); DocumentMetadataHandle meta = new DocumentMetadataHandle(); meta.getCollections().add(ENTITY); installStagingDoc("/staged.json", meta, "legacy-flow-manager/staged.json"); String pdir = "C:\\some\\crazy\\path\\to\\project"; setEnvConfig(); String flowName = "sjs-json-harmonization-flow"; LegacyFlow flow = flowManager.getFlow(ENTITY, flowName, FlowType.HARMONIZE); final String OPT_VALUE = "test-value"; Map<String, Object> options = new HashMap<String, Object>(); options.put("test-option", OPT_VALUE); Object monitor = new Object(); JobTicket jobTicket = fm.runFlow(flow, 1, 1, options, new LegacyFlowStatusListener(){ @Override public void onStatusChange(String jobId, int percentComplete, String message) { if (percentComplete == 100) { synchronized (monitor) { monitor.notify(); } } } }); synchronized (monitor) { monitor.wait(); } assertEquals(1, getFinalDocCount()); DocumentRecord doc = finalDocMgr.read("/staged.json").next(); JsonNode root = doc.getContent(new JacksonHandle()).get(); JsonNode optionNode = root.path("envelope").path("headers").path("test-option"); assertFalse(optionNode.isMissingNode()); assertEquals(OPT_VALUE, optionNode.asText()); }