ForkJoinDynamicTaskMapper implements TaskMapper { @VisibleForTesting Pair<List<WorkflowTask>, Map<String, Map<String, Object>>> getDynamicForkJoinTasksAndInput(WorkflowTask taskToSchedule, Workflow workflowInstance) throws TerminateWorkflowException { String dynamicForkJoinTaskParam = taskToSchedule.getDynamicForkJoinTasksParam(); Map<String, Object> input = parametersUtils.getTaskInput(taskToSchedule.getInputParameters(), workflowInstance, null, null); Object paramValue = input.get(dynamicForkJoinTaskParam); DynamicForkJoinTaskList dynamicForkJoinTaskList = objectMapper.convertValue(paramValue, DynamicForkJoinTaskList.class); if (dynamicForkJoinTaskList == null) { String reason = String.format("Dynamic tasks could not be created. The value of %s from task's input %s has no dynamic tasks to be scheduled", dynamicForkJoinTaskParam, input); logger.error(reason); throw new TerminateWorkflowException(reason); } Map<String, Map<String, Object>> dynamicForkJoinTasksInput = new HashMap<>(); List<WorkflowTask> dynamicForkJoinWorkflowTasks = dynamicForkJoinTaskList.getDynamicTasks().stream() .peek(dynamicForkJoinTask -> dynamicForkJoinTasksInput.put(dynamicForkJoinTask.getReferenceName(), dynamicForkJoinTask.getInput())) .map(dynamicForkJoinTask -> { WorkflowTask dynamicForkJoinWorkflowTask = new WorkflowTask(); dynamicForkJoinWorkflowTask.setTaskReferenceName(dynamicForkJoinTask.getReferenceName()); dynamicForkJoinWorkflowTask.setName(dynamicForkJoinTask.getTaskName()); dynamicForkJoinWorkflowTask.setType(dynamicForkJoinTask.getType()); if (dynamicForkJoinWorkflowTask.getTaskDefinition() == null && StringUtils.isNotBlank(dynamicForkJoinWorkflowTask.getName())) { dynamicForkJoinWorkflowTask.setTaskDefinition(metadataDAO.getTaskDef(dynamicForkJoinTask.getTaskName())); } return dynamicForkJoinWorkflowTask; }) .collect(Collectors.toCollection(LinkedList::new)); return new ImmutablePair<>(dynamicForkJoinWorkflowTasks, dynamicForkJoinTasksInput); } ForkJoinDynamicTaskMapper(ParametersUtils parametersUtils, ObjectMapper objectMapper, MetadataDAO metadataDAO); @Override List<Task> getMappedTasks(TaskMapperContext taskMapperContext); static final Logger logger; }
@Test public void getDynamicForkJoinTasksAndInput() { WorkflowTask dynamicForkJoinToSchedule = new WorkflowTask(); dynamicForkJoinToSchedule.setType(TaskType.FORK_JOIN_DYNAMIC.name()); dynamicForkJoinToSchedule.setTaskReferenceName("dynamicfanouttask"); dynamicForkJoinToSchedule.setDynamicForkJoinTasksParam("dynamicTasks"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasks", "dt1.output.dynamicTasks"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasksInput", "dt1.output.dynamicTasksInput"); DynamicForkJoinTaskList dtasks = new DynamicForkJoinTaskList(); Map<String, Object> input = new HashMap<>(); input.put("k1", "v1"); dtasks.add("junit_task_2", null, "xdt1", input); HashMap<String, Object> input2 = new HashMap<>(); input2.put("k2", "v2"); dtasks.add("junit_task_3", null, "xdt2", input2); Map<String, Object> dynamicTasksInput = new HashMap<>(); dynamicTasksInput.put("dynamicTasks", dtasks); when(parametersUtils.getTaskInput(anyMap(), any(Workflow.class), any(TaskDef.class), anyString())) .thenReturn(dynamicTasksInput); when(objectMapper.convertValue(any(), any(Class.class))).thenReturn(dtasks); Pair<List<WorkflowTask>, Map<String, Map<String, Object>>> dynamicForkJoinTasksAndInput = forkJoinDynamicTaskMapper.getDynamicForkJoinTasksAndInput(dynamicForkJoinToSchedule, new Workflow()); assertNotNull(dynamicForkJoinTasksAndInput.getLeft()); assertEquals(2,dynamicForkJoinTasksAndInput.getLeft().size()); assertEquals(2, dynamicForkJoinTasksAndInput.getRight().size()); }
@Test public void getDynamicForkJoinTasksAndInputException() { WorkflowTask dynamicForkJoinToSchedule = new WorkflowTask(); dynamicForkJoinToSchedule.setType(TaskType.FORK_JOIN_DYNAMIC.name()); dynamicForkJoinToSchedule.setTaskReferenceName("dynamicfanouttask"); dynamicForkJoinToSchedule.setDynamicForkJoinTasksParam("dynamicTasks"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasks", "dt1.output.dynamicTasks"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasksInput", "dt1.output.dynamicTasksInput"); DynamicForkJoinTaskList dtasks = new DynamicForkJoinTaskList(); Map<String, Object> input = new HashMap<>(); input.put("k1", "v1"); dtasks.add("junit_task_2", null, "xdt1", input); HashMap<String, Object> input2 = new HashMap<>(); input2.put("k2", "v2"); dtasks.add("junit_task_3", null, "xdt2", input2); Map<String, Object> dynamicTasksInput = new HashMap<>(); dynamicTasksInput.put("dynamicTasks", dtasks); when(parametersUtils.getTaskInput(anyMap(), any(Workflow.class), any(TaskDef.class), anyString())) .thenReturn(dynamicTasksInput); when(objectMapper.convertValue(any(), any(Class.class))).thenReturn(null); expectedException.expect(TerminateWorkflowException.class); forkJoinDynamicTaskMapper.getDynamicForkJoinTasksAndInput(dynamicForkJoinToSchedule, new Workflow()); }