EntitiesController { @RequestMapping(value = "/entities/{entityName}/flows/harmonize/{flowName}/run", method = RequestMethod.POST) @ResponseBody public ResponseEntity<?> runHarmonizeFlow( @PathVariable String entityName, @PathVariable String flowName, @RequestBody JsonNode json) { int batchSize = json.get("batchSize").asInt(); int threadCount = json.get("threadCount").asInt(); Map<String, Object> options = new HashMap<>(); if(json.get("options") != null && json.get("options").size() > 0) { Iterator<Map.Entry<String, JsonNode>> optionIter = json.get("options").fields(); Map.Entry<String, JsonNode> current; while (optionIter.hasNext()) { current = optionIter.next(); options.put(current.getKey(), current.getValue()); } } ResponseEntity<?> resp; LegacyFlow flow = legacyFlowManagerService.getServerFlow(entityName, flowName, FlowType.HARMONIZE); if (flow == null) { resp = new ResponseEntity<>(HttpStatus.CONFLICT); } else { legacyFlowManagerService.runFlow(flow, batchSize, threadCount, options, (jobId, percentComplete, message) -> { template.convertAndSend("/topic/flow-status", new JobStatusMessage(jobId, percentComplete, message, FlowType.HARMONIZE.toString())); }); resp = new ResponseEntity<>(HttpStatus.OK); } return resp; }  @RequestMapping(value = "/entities/create", method = RequestMethod.POST) @ResponseBody EntityModel createEntity(@RequestBody EntityModel newEntity); @RequestMapping(value = "/entities/", method = RequestMethod.GET) @ResponseBody Collection<EntityModel> getEntities(); @RequestMapping(value = "/entities/", method = RequestMethod.POST) @ResponseBody ResponseEntity<?> saveEntities(@RequestBody List<EntityModel> entities); @RequestMapping(value = "/entities/ui/", method = RequestMethod.POST) @ResponseBody ResponseEntity<?> saveEntitiesUiState(@RequestBody List<EntityModel> entities); @RequestMapping(value = "/entities/{entityName}", method = RequestMethod.PUT) @ResponseBody EntityModel saveEntity(@RequestBody EntityModel entity); @RequestMapping(value = "/entities/{entityName}", method = RequestMethod.GET) @ResponseBody EntityModel getEntity(@PathVariable String entityName, @RequestParam(required = false)Boolean extendSubEntities); @RequestMapping(value = "/entities/{entityName}", method = RequestMethod.DELETE) @ResponseBody ResponseEntity<?> deleteEntity(@PathVariable String entityName); @RequestMapping(value = "/entities/{entityName}/flows/{flowType}", method = RequestMethod.POST) @ResponseBody FlowModel createFlow(
            @PathVariable String entityName,
            @PathVariable FlowType flowType,
            @RequestBody FlowModel newFlow); @RequestMapping(value = "/entities/{entityName}/flows/{flowName}/{flowType}", method = RequestMethod.DELETE) @ResponseBody ResponseEntity<?> deleteFlow(
        @PathVariable String entityName,
        @PathVariable String flowName,
        @PathVariable FlowType flowType); @RequestMapping(value = "/entities/{entityName}/flows/harmonize/{flowName}/run", method = RequestMethod.POST) @ResponseBody ResponseEntity<?> runHarmonizeFlow(
            @PathVariable String entityName,
            @PathVariable String flowName,
            @RequestBody JsonNode json); @RequestMapping( value = "/plugin/save", method = RequestMethod.POST ) @ResponseBody String saveFlowPlugin(
        @RequestBody PluginModel plugin); @RequestMapping( value = "/entities/{entityName}/flows/{flowType}/{flowName}/plugin/validate", method = RequestMethod.POST ) @ResponseBody JsonNode validateFlowPlugin(
        @PathVariable String entityName,
        @PathVariable FlowType flowType,
        @PathVariable String flowName,
        @RequestBody PluginModel plugin); @RequestMapping(value = "/entities/{entityName}/flows/harmonize/{flowName}/save-harmonize-options", method = RequestMethod.POST) @ResponseBody ResponseEntity<?> saveHarmonizeFlowOptions(
        @PathVariable String entityName,
        @PathVariable String flowName,
        @RequestBody JsonNode json); @RequestMapping(value = "/entities/{entityName}/flows/input/{flowName}/save-input-options", method = RequestMethod.POST) @ResponseBody ResponseEntity<?> saveInputFlowOptions(
        @PathVariable String entityName,
        @PathVariable String flowName,
        @RequestBody JsonNode json); @RequestMapping(value = "/entities/{entityName}/flows/input/{flowName}/run", method = RequestMethod.POST) ResponseEntity<?> runInputFlow(
            @PathVariable String entityName,
            @PathVariable String flowName,
            @RequestBody JsonNode json); @RequestMapping(value = "/entities/{entityName}/flows/{flowName}/cancel/{jobId}", method = RequestMethod.DELETE) ResponseEntity<?> cancelFlow(
            @PathVariable String entityName,
            @PathVariable String flowName,
            @PathVariable String jobId); @RequestMapping( value = "/entities/{entityName}/flows/input/{flowName}/run", method = RequestMethod.GET, produces="application/json" ) @ResponseBody Map<String, Object> getInputFlowOptions(
            @PathVariable String entityName,
            @PathVariable String flowName);  }
@Test public void runHarmonizeNoOptions() throws IOException, InterruptedException { deleteProjectDir(); createProjectDir(); Path projectDir = Paths.get(".", PROJECT_PATH); scaffolding.createLegacyFlow(ENTITY, "sjs-json-harmonization-flow", FlowType.HARMONIZE, CodeFormat.JAVASCRIPT, DataFormat.JSON, false); Path harmonizeDir = projectDir.resolve("plugins/entities/" + ENTITY + "/harmonize"); InputStream inputStream = getResourceStream("legacy-flow-manager/sjs-harmonize-flow/headers.sjs"); FileUtil.copy(inputStream, harmonizeDir.resolve("sjs-json-harmonization-flow/headers.sjs").toFile()); IOUtils.closeQuietly(inputStream); installUserModules(getDataHubAdminConfig(), true); DocumentMetadataHandle meta = new DocumentMetadataHandle(); meta.getCollections().add(ENTITY); installStagingDoc("/staged.json", meta, "legacy-flow-manager/staged.json"); setEnvConfig(); ObjectMapper mapper = new ObjectMapper(); JsonNode body = mapper.readTree("{\"batchSize\":1, \"threadCount\": 1}"); ResponseEntity<?> responseEntity = ec.runHarmonizeFlow(ENTITY, "sjs-json-harmonization-flow", body); assertEquals(HttpStatus.OK, responseEntity.getStatusCode()); Thread.sleep(3000); DocumentRecord doc = finalDocMgr.read("/staged.json").next(); JsonNode root = doc.getContent(new JacksonHandle()).get(); JsonNode env = root.path("envelope"); JsonNode headers = env.path("headers"); JsonNode optionNode = headers.path("test-option"); assertTrue(optionNode.isMissingNode()); }
@Test public void runHarmonizeFlowWithOptions() throws IOException, InterruptedException { deleteProjectDir(); createProjectDir(); Path projectDir = Paths.get(".", PROJECT_PATH); scaffolding.createLegacyFlow(ENTITY, "sjs-json-harmonization-flow", FlowType.HARMONIZE, CodeFormat.JAVASCRIPT, DataFormat.JSON, false); Path harmonizeDir = projectDir.resolve("plugins/entities/" + ENTITY + "/harmonize"); InputStream inputStream = getResourceStream("legacy-flow-manager/sjs-harmonize-flow/headers.sjs"); FileUtil.copy(inputStream, harmonizeDir.resolve("sjs-json-harmonization-flow/headers.sjs").toFile()); IOUtils.closeQuietly(inputStream); installUserModules(getDataHubAdminConfig(), true); DocumentMetadataHandle meta = new DocumentMetadataHandle(); meta.getCollections().add(ENTITY); installStagingDoc("/staged.json", meta, "legacy-flow-manager/staged.json"); setEnvConfig(); final String OPT_VALUE = "test-value"; ObjectMapper mapper = new ObjectMapper(); JsonNode body = mapper.readTree("{\"batchSize\":1, \"threadCount\": 1, \"options\": {\"test-option\": \"" + OPT_VALUE + "\"}}"); ResponseEntity<?> responseEntity = ec.runHarmonizeFlow(ENTITY, "sjs-json-harmonization-flow", body); assertEquals(HttpStatus.OK, responseEntity.getStatusCode()); Thread.sleep(3000); DocumentRecord doc = finalDocMgr.read("/staged.json").next(); JsonNode root = doc.getContent(new JacksonHandle()).get(); JsonNode env = root.path("envelope"); JsonNode headers = env.path("headers"); JsonNode optionNode = headers.path("test-option"); assertFalse(optionNode.isMissingNode()); assertEquals(OPT_VALUE, optionNode.asText()); }