SimpleActionProcessor implements ActionProcessor { private Map<String, Object> startWorkflow(Action action, Object payload, String event, String messageId) { StartWorkflow params = action.getStart_workflow(); Map<String, Object> output = new HashMap<>(); try { Map<String, Object> inputParams = params.getInput(); Map<String, Object> workflowInput = parametersUtils.replace(inputParams, payload); Map<String, Object> paramsMap = new HashMap<>(); Optional.ofNullable(params.getCorrelationId()) .ifPresent(value -> paramsMap.put("correlationId", value)); Map<String, Object> replaced = parametersUtils.replace(paramsMap, payload); workflowInput.put("conductor.event.messageId", messageId); workflowInput.put("conductor.event.name", event); String workflowId = executor.startWorkflow(params.getName(), params.getVersion(), Optional.ofNullable(replaced.get("correlationId")).map(Object::toString) .orElse(params.getCorrelationId()), workflowInput, null, event, params.getTaskToDomain()); output.put("workflowId", workflowId); logger.debug("Started workflow: {}/{}/{} for event: {} for message:{}", params.getName(), params.getVersion(), workflowId, event, messageId); } catch (RuntimeException e) { Monitors.recordEventActionError(action.getAction().name(), params.getName(), event); logger.error("Error starting workflow: {}, version: {}, for event: {} for message: {}", params.getName(), params.getVersion(), event, messageId, e); output.put("error", e.getMessage()); throw e; } return output; } @Inject  SimpleActionProcessor(WorkflowExecutor executor, ParametersUtils parametersUtils, JsonUtils jsonUtils); Map<String, Object> execute(Action action, Object payloadObject, String event, String messageId);  }
@SuppressWarnings("unchecked") @Test public void testStartWorkflow() throws Exception { StartWorkflow startWorkflow = new StartWorkflow(); startWorkflow.setName("testWorkflow"); startWorkflow.getInput().put("testInput", "${testId}"); Map<String, String> taskToDomain = new HashMap<>(); taskToDomain.put("*", "dev"); startWorkflow.setTaskToDomain(taskToDomain); Action action = new Action(); action.setAction(Type.start_workflow); action.setStart_workflow(startWorkflow); Object payload = objectMapper.readValue("{\"testId\":\"test_1\"}", Object.class); WorkflowDef workflowDef = new WorkflowDef(); workflowDef.setName("testWorkflow"); workflowDef.setVersion(1); when(workflowExecutor.startWorkflow(eq("testWorkflow"), eq(null), any(), any(), any(), eq("testEvent"), anyMap())) .thenReturn("workflow_1"); Map<String, Object> output = actionProcessor.execute(action, payload, "testEvent", "testMessage"); assertNotNull(output); assertEquals("workflow_1", output.get("workflowId")); ArgumentCaptor<String> correlationIdCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Map> inputParamCaptor = ArgumentCaptor.forClass(Map.class); ArgumentCaptor<Map> taskToDomainCaptor = ArgumentCaptor.forClass(Map.class); verify(workflowExecutor).startWorkflow(eq("testWorkflow"), eq(null), correlationIdCaptor.capture(), inputParamCaptor.capture(), any(), eq("testEvent"), taskToDomainCaptor.capture()); assertEquals("test_1", inputParamCaptor.getValue().get("testInput")); assertNull(correlationIdCaptor.getValue()); assertEquals("testMessage", inputParamCaptor.getValue().get("conductor.event.messageId")); assertEquals("testEvent", inputParamCaptor.getValue().get("conductor.event.name")); assertEquals(taskToDomain, taskToDomainCaptor.getValue()); }