EventTaskMapper implements TaskMapper { @Override public List<Task> getMappedTasks(TaskMapperContext taskMapperContext) { logger.debug("TaskMapperContext {} in EventTaskMapper", taskMapperContext); WorkflowTask taskToSchedule = taskMapperContext.getTaskToSchedule(); Workflow workflowInstance = taskMapperContext.getWorkflowInstance(); String taskId = taskMapperContext.getTaskId(); taskToSchedule.getInputParameters().put("sink", taskToSchedule.getSink()); taskToSchedule.getInputParameters().put("asyncComplete", taskToSchedule.isAsyncComplete()); Map<String, Object> eventTaskInput = parametersUtils.getTaskInputV2(taskToSchedule.getInputParameters(), workflowInstance, taskId, null); String sink = (String) eventTaskInput.get("sink"); Boolean asynComplete = (Boolean) eventTaskInput.get("asyncComplete"); Task eventTask = new Task(); eventTask.setTaskType(Event.NAME); eventTask.setTaskDefName(taskToSchedule.getName()); eventTask.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); eventTask.setWorkflowInstanceId(workflowInstance.getWorkflowId()); eventTask.setWorkflowType(workflowInstance.getWorkflowName()); eventTask.setCorrelationId(workflowInstance.getCorrelationId()); eventTask.setScheduledTime(System.currentTimeMillis()); eventTask.setInputData(eventTaskInput); eventTask.getInputData().put("sink", sink); eventTask.getInputData().put("asyncComplete", asynComplete); eventTask.setTaskId(taskId); eventTask.setStatus(Task.Status.SCHEDULED); eventTask.setWorkflowPriority(workflowInstance.getPriority()); eventTask.setWorkflowTask(taskToSchedule); return Collections.singletonList(eventTask); } EventTaskMapper(ParametersUtils parametersUtils); @Override List<Task> getMappedTasks(TaskMapperContext taskMapperContext); static final Logger logger; }
@Test public void getMappedTasks() throws Exception { ParametersUtils parametersUtils = Mockito.mock(ParametersUtils.class); EventTaskMapper eventTaskMapper = new EventTaskMapper(parametersUtils); WorkflowTask taskToBeScheduled = new WorkflowTask(); taskToBeScheduled.setSink("SQSSINK"); String taskId = IDGenerator.generate(); Map<String, Object> eventTaskInput = new HashMap<>(); eventTaskInput.put("sink","SQSSINK"); when(parametersUtils.getTaskInput(anyMap(), any(Workflow.class), any(TaskDef.class), anyString())).thenReturn(eventTaskInput); WorkflowDef wd = new WorkflowDef(); Workflow w = new Workflow(); w.setWorkflowDefinition(wd); TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder() .withWorkflowDefinition(wd) .withWorkflowInstance(w) .withTaskDefinition(new TaskDef()) .withTaskToSchedule(taskToBeScheduled) .withRetryCount(0) .withTaskId(taskId) .build(); List<Task> mappedTasks = eventTaskMapper.getMappedTasks(taskMapperContext); assertEquals(1, mappedTasks.size()); Task eventTask = mappedTasks.get(0); assertEquals(taskId, eventTask.getTaskId()); }