SubWorkflow extends WorkflowSystemTask { @SuppressWarnings("unchecked") @Override public void start(Workflow workflow, Task task, WorkflowExecutor provider) { Map<String, Object> input = task.getInputData(); String name = input.get("subWorkflowName").toString(); int version = (int) input.get("subWorkflowVersion"); WorkflowDef workflowDefinition = null; if (input.get("subWorkflowDefinition") != null) { workflowDefinition = objectMapperSupplier.get().convertValue(input.get("subWorkflowDefinition"), WorkflowDef.class); } Map taskToDomain = workflow.getTaskToDomain(); if (input.get("subWorkflowTaskToDomain") instanceof Map) { taskToDomain = (Map) input.get("subWorkflowTaskToDomain"); } Map<String, Object> wfInput = (Map<String, Object>) input.get("workflowInput"); if (wfInput == null || wfInput.isEmpty()) { wfInput = input; } String correlationId = workflow.getCorrelationId(); try { String subWorkflowId = null; if (workflowDefinition != null) { subWorkflowId = provider.startWorkflow( workflowDefinition, wfInput, null, correlationId, 0, workflow.getWorkflowId(), task.getTaskId(), null, taskToDomain); } else { subWorkflowId = provider.startWorkflow( name, version, wfInput, null, correlationId, workflow.getWorkflowId(), task.getTaskId(), null, taskToDomain); } task.setSubWorkflowId(subWorkflowId); task.getOutputData().put(SUB_WORKFLOW_ID, subWorkflowId); Workflow subWorkflow = provider.getWorkflow(subWorkflowId, false); updateTaskStatus(subWorkflow.getStatus(), task); } catch (Exception e) { task.setStatus(Status.FAILED); task.setReasonForIncompletion(e.getMessage()); logger.error("Error starting sub workflow: {} from workflow: {}", name, workflow.getWorkflowId(), e); } } @VisibleForTesting  SubWorkflow(); @Inject  SubWorkflow(ObjectMapper objectMapper); @SuppressWarnings("unchecked") @Override void start(Workflow workflow, Task task, WorkflowExecutor provider); @Override boolean execute(Workflow workflow, Task task, WorkflowExecutor provider); @Override void cancel(Workflow workflow, Task task, WorkflowExecutor provider); @Override boolean isAsync(); @Override boolean isAsyncComplete(Task task); static final String NAME; }
@Test public void testStartSubWorkflow() { WorkflowDef workflowDef = new WorkflowDef(); Workflow workflowInstance = new Workflow(); workflowInstance.setWorkflowDefinition(workflowDef); Task task = new Task(); task.setOutputData(new HashMap<>()); Map<String, Object> inputData = new HashMap<>(); inputData.put("subWorkflowName", "UnitWorkFlow"); inputData.put("subWorkflowVersion", 3); task.setInputData(inputData); String workflowId = "workflow_1"; Workflow workflow = new Workflow(); workflow.setWorkflowId(workflowId); when(workflowExecutor.startWorkflow(eq("UnitWorkFlow"), eq(3), eq(inputData), eq(null), any(), any(), any(), eq(null), any())) .thenReturn(workflowId); when(workflowExecutor.getWorkflow(anyString(), eq(false))).thenReturn(workflow); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); assertEquals(Task.Status.IN_PROGRESS, task.getStatus()); workflow.setStatus(Workflow.WorkflowStatus.TERMINATED); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); assertEquals(Task.Status.FAILED, task.getStatus()); workflow.setStatus(Workflow.WorkflowStatus.COMPLETED); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); assertEquals(Task.Status.COMPLETED, task.getStatus()); }
@Test public void testStartSubWorkflowWithEmptyWorkflowInput() { WorkflowDef workflowDef = new WorkflowDef(); Workflow workflowInstance = new Workflow(); workflowInstance.setWorkflowDefinition(workflowDef); Task task = new Task(); task.setOutputData(new HashMap<>()); Map<String, Object> inputData = new HashMap<>(); inputData.put("subWorkflowName", "UnitWorkFlow"); inputData.put("subWorkflowVersion", 3); Map<String, Object> workflowInput = new HashMap<>(); inputData.put("workflowInput", workflowInput); task.setInputData(inputData); when(workflowExecutor.startWorkflow(eq("UnitWorkFlow"), eq(3), eq(inputData), eq(null), any(), any(), any(), eq(null), any())) .thenReturn("workflow_1"); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); }
@Test public void testStartSubWorkflowWithWorkflowInput() { WorkflowDef workflowDef = new WorkflowDef(); Workflow workflowInstance = new Workflow(); workflowInstance.setWorkflowDefinition(workflowDef); Task task = new Task(); task.setOutputData(new HashMap<>()); Map<String, Object> inputData = new HashMap<>(); inputData.put("subWorkflowName", "UnitWorkFlow"); inputData.put("subWorkflowVersion", 3); Map<String, Object> workflowInput = new HashMap<>(); workflowInput.put("test", "value"); inputData.put("workflowInput", workflowInput); task.setInputData(inputData); when(workflowExecutor.startWorkflow(eq("UnitWorkFlow"), eq(3), eq(workflowInput), eq(null), any(), any(), any(), eq(null), any())) .thenReturn("workflow_1"); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); }
@Test public void testStartSubWorkflowTaskToDomain() { WorkflowDef workflowDef = new WorkflowDef(); Workflow workflowInstance = new Workflow(); workflowInstance.setWorkflowDefinition(workflowDef); Map<String, String> taskToDomain = new HashMap<String, String>() {{put("*", "unittest"); }}; Task task = new Task(); task.setOutputData(new HashMap<>()); Map<String, Object> inputData = new HashMap<>(); inputData.put("subWorkflowName", "UnitWorkFlow"); inputData.put("subWorkflowVersion", 2); inputData.put("subWorkflowTaskToDomain", taskToDomain); task.setInputData(inputData); when(workflowExecutor.startWorkflow(eq("UnitWorkFlow"), eq(2), eq(inputData), eq(null), any(), any(), any(), eq(null), eq(taskToDomain))) .thenReturn("workflow_1"); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); }
@Test public void testStartSubWorkflowWithSubWorkflowDefinition() { WorkflowDef workflowDef = new WorkflowDef(); Workflow workflowInstance = new Workflow(); workflowInstance.setWorkflowDefinition(workflowDef); WorkflowDef subWorkflowDef = new WorkflowDef(); subWorkflowDef.setName("subWorkflow_1"); Task task = new Task(); task.setOutputData(new HashMap<>()); Map<String, Object> inputData = new HashMap<>(); inputData.put("subWorkflowName", "UnitWorkFlow"); inputData.put("subWorkflowVersion", 2); inputData.put("subWorkflowDefinition", subWorkflowDef); task.setInputData(inputData); when(workflowExecutor.startWorkflow(eq(subWorkflowDef), eq(inputData), eq(null), any(), eq(0), any(), any(), eq(null), any())) .thenReturn("workflow_1"); subWorkflow.start(workflowInstance, task, workflowExecutor); assertEquals("workflow_1", task.getSubWorkflowId()); }