ForkJoinDynamicTaskMapper implements TaskMapper { @SuppressWarnings("unchecked") @VisibleForTesting Pair<List<WorkflowTask>, Map<String, Map<String, Object>>> getDynamicForkTasksAndInput(WorkflowTask taskToSchedule, Workflow workflowInstance, String dynamicForkTaskParam) throws TerminateWorkflowException { Map<String, Object> input = parametersUtils.getTaskInput(taskToSchedule.getInputParameters(), workflowInstance, null, null); Object dynamicForkTasksJson = input.get(dynamicForkTaskParam); List<WorkflowTask> dynamicForkWorkflowTasks = objectMapper.convertValue(dynamicForkTasksJson, ListOfWorkflowTasks); if(dynamicForkWorkflowTasks == null) { dynamicForkWorkflowTasks = new ArrayList<WorkflowTask>(); } for (WorkflowTask workflowTask : dynamicForkWorkflowTasks) { if ((workflowTask.getTaskDefinition() == null) && StringUtils.isNotBlank(workflowTask.getName())) { workflowTask.setTaskDefinition(metadataDAO.getTaskDef(workflowTask.getName())); } } Object dynamicForkTasksInput = input.get(taskToSchedule.getDynamicForkTasksInputParamName()); if (!(dynamicForkTasksInput instanceof Map)) { throw new TerminateWorkflowException("Input to the dynamically forked tasks is not a map -> expecting a map of K,V but found " + dynamicForkTasksInput); } return new ImmutablePair<>(dynamicForkWorkflowTasks, (Map<String, Map<String, Object>>) dynamicForkTasksInput); } ForkJoinDynamicTaskMapper(ParametersUtils parametersUtils, ObjectMapper objectMapper, MetadataDAO metadataDAO); @Override List<Task> getMappedTasks(TaskMapperContext taskMapperContext); static final Logger logger; }
@Test public void getDynamicForkTasksAndInput() { WorkflowTask dynamicForkJoinToSchedule = new WorkflowTask(); dynamicForkJoinToSchedule.setType(TaskType.FORK_JOIN_DYNAMIC.name()); dynamicForkJoinToSchedule.setTaskReferenceName("dynamicfanouttask"); dynamicForkJoinToSchedule.setDynamicForkTasksParam("dynamicTasks"); dynamicForkJoinToSchedule.setDynamicForkTasksInputParamName("dynamicTasksInput"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasks", "dt1.output.dynamicTasks"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasksInput", "dt1.output.dynamicTasksInput"); Map<String, Object> input1 = new HashMap<>(); input1.put("k1", "v1"); WorkflowTask wt2 = new WorkflowTask(); wt2.setName("junit_task_2"); wt2.setTaskReferenceName("xdt1"); Map<String, Object> input2 = new HashMap<>(); input2.put("k2", "v2"); WorkflowTask wt3 = new WorkflowTask(); wt3.setName("junit_task_3"); wt3.setTaskReferenceName("xdt2"); HashMap<String, Object> dynamicTasksInput = new HashMap<>(); dynamicTasksInput.put("xdt1", input1); dynamicTasksInput.put("xdt2", input2); dynamicTasksInput.put("dynamicTasks", Arrays.asList(wt2, wt3)); dynamicTasksInput.put("dynamicTasksInput", dynamicTasksInput); when(parametersUtils.getTaskInput(anyMap(), any(Workflow.class), any(), any())) .thenReturn(dynamicTasksInput); when(objectMapper.convertValue(any(), any(TypeReference.class))).thenReturn(Arrays.asList(wt2, wt3)); Pair<List<WorkflowTask>, Map<String, Map<String, Object>>> dynamicTasks = forkJoinDynamicTaskMapper.getDynamicForkTasksAndInput(dynamicForkJoinToSchedule, new Workflow(), "dynamicTasks"); assertNotNull(dynamicTasks.getLeft()); }
@Test public void getDynamicForkTasksAndInputException() { WorkflowTask dynamicForkJoinToSchedule = new WorkflowTask(); dynamicForkJoinToSchedule.setType(TaskType.FORK_JOIN_DYNAMIC.name()); dynamicForkJoinToSchedule.setTaskReferenceName("dynamicfanouttask"); dynamicForkJoinToSchedule.setDynamicForkTasksParam("dynamicTasks"); dynamicForkJoinToSchedule.setDynamicForkTasksInputParamName("dynamicTasksInput"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasks", "dt1.output.dynamicTasks"); dynamicForkJoinToSchedule.getInputParameters().put("dynamicTasksInput", "dt1.output.dynamicTasksInput"); Map<String, Object> input1 = new HashMap<>(); input1.put("k1", "v1"); WorkflowTask wt2 = new WorkflowTask(); wt2.setName("junit_task_2"); wt2.setTaskReferenceName("xdt1"); Map<String, Object> input2 = new HashMap<>(); input2.put("k2", "v2"); WorkflowTask wt3 = new WorkflowTask(); wt3.setName("junit_task_3"); wt3.setTaskReferenceName("xdt2"); HashMap<String, Object> dynamicTasksInput = new HashMap<>(); dynamicTasksInput.put("xdt1", input1); dynamicTasksInput.put("xdt2", input2); dynamicTasksInput.put("dynamicTasks", Arrays.asList(wt2, wt3)); dynamicTasksInput.put("dynamicTasksInput", null); when(parametersUtils.getTaskInput(anyMap(), any(Workflow.class), any(), any())) .thenReturn(dynamicTasksInput); when(objectMapper.convertValue(any(), any(TypeReference.class))).thenReturn(Arrays.asList(wt2, wt3)); expectedException.expect(TerminateWorkflowException.class); forkJoinDynamicTaskMapper.getDynamicForkTasksAndInput(dynamicForkJoinToSchedule, new Workflow(), "dynamicTasks"); }