DynamicTaskMapper implements TaskMapper { @Override public List<Task> getMappedTasks(TaskMapperContext taskMapperContext) throws TerminateWorkflowException { logger.debug("TaskMapperContext {} in DynamicTaskMapper", taskMapperContext); WorkflowTask taskToSchedule = taskMapperContext.getTaskToSchedule(); Map<String, Object> taskInput = taskMapperContext.getTaskInput(); Workflow workflowInstance = taskMapperContext.getWorkflowInstance(); int retryCount = taskMapperContext.getRetryCount(); String retriedTaskId = taskMapperContext.getRetryTaskId(); String taskNameParam = taskToSchedule.getDynamicTaskNameParam(); String taskName = getDynamicTaskName(taskInput, taskNameParam); taskToSchedule.setName(taskName); TaskDef taskDefinition = getDynamicTaskDefinition(taskToSchedule); taskToSchedule.setTaskDefinition(taskDefinition); Map<String, Object> input = parametersUtils.getTaskInput(taskToSchedule.getInputParameters(), workflowInstance, taskDefinition, taskMapperContext.getTaskId()); Task dynamicTask = new Task(); dynamicTask.setStartDelayInSeconds(taskToSchedule.getStartDelay()); dynamicTask.setTaskId(taskMapperContext.getTaskId()); dynamicTask.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); dynamicTask.setInputData(input); dynamicTask.setWorkflowInstanceId(workflowInstance.getWorkflowId()); dynamicTask.setWorkflowType(workflowInstance.getWorkflowName()); dynamicTask.setStatus(Task.Status.SCHEDULED); dynamicTask.setTaskType(taskToSchedule.getType()); dynamicTask.setTaskDefName(taskToSchedule.getName()); dynamicTask.setCorrelationId(workflowInstance.getCorrelationId()); dynamicTask.setScheduledTime(System.currentTimeMillis()); dynamicTask.setRetryCount(retryCount); dynamicTask.setCallbackAfterSeconds(taskToSchedule.getStartDelay()); dynamicTask.setResponseTimeoutSeconds(taskDefinition.getResponseTimeoutSeconds()); dynamicTask.setWorkflowTask(taskToSchedule); dynamicTask.setTaskType(taskName); dynamicTask.setRetriedTaskId(retriedTaskId); dynamicTask.setWorkflowPriority(workflowInstance.getPriority()); return Collections.singletonList(dynamicTask); } DynamicTaskMapper(ParametersUtils parametersUtils, MetadataDAO metadataDAO); @Override List<Task> getMappedTasks(TaskMapperContext taskMapperContext);  }
@SuppressWarnings("unchecked") @Test public void getMappedTasks() { WorkflowTask workflowTask = new WorkflowTask(); workflowTask.setName("DynoTask"); workflowTask.setDynamicTaskNameParam("dynamicTaskName"); TaskDef taskDef = new TaskDef(); taskDef.setName("DynoTask"); workflowTask.setTaskDefinition(taskDef); Map<String, Object> taskInput = new HashMap<>(); taskInput.put("dynamicTaskName", "DynoTask"); when(parametersUtils.getTaskInput(anyMap(), any(Workflow.class), any(TaskDef.class), anyString())).thenReturn(taskInput); String taskId = IDGenerator.generate(); Workflow workflow = new Workflow(); WorkflowDef workflowDef = new WorkflowDef(); workflow.setWorkflowDefinition(workflowDef); TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder() .withWorkflowInstance(workflow) .withWorkflowDefinition(workflowDef) .withTaskDefinition(workflowTask.getTaskDefinition()) .withTaskToSchedule(workflowTask) .withTaskInput(taskInput) .withRetryCount(0) .withTaskId(taskId) .build(); when(metadataDAO.getTaskDef("DynoTask")).thenReturn(new TaskDef()); List<Task> mappedTasks = dynamicTaskMapper.getMappedTasks(taskMapperContext); assertEquals(1, mappedTasks.size()); Task dynamicTask = mappedTasks.get(0); assertEquals(taskId, dynamicTask.getTaskId()); }