SubWorkflow extends WorkflowSystemTask { @Override public boolean execute(Workflow workflow, Task task, WorkflowExecutor provider) { String workflowId = task.getSubWorkflowId(); if(StringUtils.isEmpty(workflowId)) { return false; } Workflow subWorkflow = provider.getWorkflow(workflowId, false); WorkflowStatus subWorkflowStatus = subWorkflow.getStatus(); if(!subWorkflowStatus.isTerminal()){ return false; } if (subWorkflow.getExternalOutputPayloadStoragePath() != null) { task.setExternalOutputPayloadStoragePath(subWorkflow.getExternalOutputPayloadStoragePath()); } else { task.getOutputData().putAll(subWorkflow.getOutput()); } if (!subWorkflowStatus.isSuccessful()) { task.setReasonForIncompletion(subWorkflow.getReasonForIncompletion()); } updateTaskStatus(subWorkflowStatus, task); return true; } @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 testExecuteSubWorkflowWithoutId() { 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", 2); task.setInputData(inputData); when(workflowExecutor.startWorkflow(eq("UnitWorkFlow"), eq(2), eq(inputData), eq(null), any(), any(), any(), eq(null), eq(null))) .thenReturn("workflow_1"); assertFalse(subWorkflow.execute(workflowInstance, task, workflowExecutor)); }
@Test public void testExecuteWorkflowStatus() { WorkflowDef workflowDef = new WorkflowDef(); Workflow workflowInstance = new Workflow(); Workflow subWorkflowInstance = new Workflow(); workflowInstance.setWorkflowDefinition(workflowDef); Map<String, String> taskToDomain = new HashMap<String, String>() {{put("*", "unittest"); }}; Task task = new Task(); Map<String, Object> outputData = new HashMap<>(); task.setOutputData(outputData); task.setSubWorkflowId("sub-workflow-id"); 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"); when(workflowExecutor.getWorkflow(eq("sub-workflow-id"), eq(false))) .thenReturn(subWorkflowInstance); subWorkflowInstance.setStatus(Workflow.WorkflowStatus.RUNNING); assertFalse(subWorkflow.execute(workflowInstance, task, workflowExecutor)); assertNull(task.getStatus()); assertNull(task.getReasonForIncompletion()); subWorkflowInstance.setStatus(Workflow.WorkflowStatus.PAUSED); assertFalse(subWorkflow.execute(workflowInstance, task, workflowExecutor)); assertNull(task.getStatus()); assertNull(task.getReasonForIncompletion()); subWorkflowInstance.setStatus(Workflow.WorkflowStatus.COMPLETED); assertTrue(subWorkflow.execute(workflowInstance, task, workflowExecutor)); assertEquals(Task.Status.COMPLETED, task.getStatus()); assertNull(task.getReasonForIncompletion()); subWorkflowInstance.setStatus(Workflow.WorkflowStatus.FAILED); subWorkflowInstance.setReasonForIncompletion("unit1"); assertTrue(subWorkflow.execute(workflowInstance, task, workflowExecutor)); assertEquals(Task.Status.FAILED, task.getStatus()); assertEquals("unit1", task.getReasonForIncompletion()); subWorkflowInstance.setStatus(Workflow.WorkflowStatus.TIMED_OUT); subWorkflowInstance.setReasonForIncompletion("unit2"); assertTrue(subWorkflow.execute(workflowInstance, task, workflowExecutor)); assertEquals(Task.Status.TIMED_OUT, task.getStatus()); assertEquals("unit2", task.getReasonForIncompletion()); subWorkflowInstance.setStatus(Workflow.WorkflowStatus.TERMINATED); subWorkflowInstance.setReasonForIncompletion("unit3"); assertTrue(subWorkflow.execute(workflowInstance, task, workflowExecutor)); assertEquals(Task.Status.FAILED, task.getStatus()); assertEquals("unit3", task.getReasonForIncompletion()); }