LegacyFlowManagerService { public void runMlcp(LegacyFlow flow, JsonNode json, LegacyFlowStatusListener statusListener) { String mlcpPath = json.get("mlcpPath").textValue(); MlcpRunner runner = new MlcpRunner(mlcpPath, "com.marklogic.contentpump.ContentPump", hubConfig, flow, hubConfig.newStagingClient(), json.get("mlcpOptions"), statusListener); runner.start(); }  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 @Disabled public void runMlcp() throws IOException, InterruptedException, JSONException { clearDatabases(HubConfig.DEFAULT_STAGING_NAME, HubConfig.DEFAULT_FINAL_NAME, HubConfig.DEFAULT_JOB_NAME); String flowName = "sjs-json-input-flow"; LegacyFlow flow = flowManager.getFlow(ENTITY, flowName, FlowType.INPUT); ObjectMapper objectMapper = new ObjectMapper(); String inputPath = getResourceFile("legacy-flow-manager/input.json").getAbsolutePath(); String basePath = getResourceFile("legacy-flow-manager").getAbsolutePath(); JsonNode mlcpOptions = objectMapper.readTree( "{" + "\"input_file_path\":\"" + inputPath.replace("\\", "\\\\\\\\") + "\"," + "\"input_file_type\":\"\\\"documents\\\"\"," + "\"output_collections\":\"\\\"" + ENTITY + "\\\"\"," + "\"output_permissions\":\"\\\"rest-reader,read,rest-writer,update\\\"\"," + "\"output_uri_replace\":\"\\\"" + basePath.replace("\\", "/").replaceAll("^([A-Za-z]):", "/$1:") + ",''\\\"\"," + "\"document_type\":\"\\\"json\\\"\"," + "\"transform_module\":\"\\\"/data-hub/4/transforms/mlcp-flow-transform.sjs\\\"\"," + "\"transform_namespace\":\"\\\"http: "\"transform_param\":\"\\\"entity-name=" + ENTITY + ",flow-name=" + flowName + "\\\"\"" + "}"); LegacyFlowStatusListener flowStatusListener = (jobId, percentComplete, message) -> { logger.error(message); }; MlcpRunner mlcpRunner = new MlcpRunner(null, "com.marklogic.hub.util.MlcpMain", runAsFlowOperator(), flow, stagingClient, mlcpOptions, flowStatusListener); mlcpRunner.start(); mlcpRunner.join(); assertEquals(1, getStagingDocCount()); String expected = getResource("legacy-flow-manager/final.json"); String actual = stagingDocMgr.read("/input.json").next().getContent(new StringHandle()).get(); JSONAssert.assertEquals(expected, actual, false); }