SimpleTaskMapper implements TaskMapper { @Override public List<Task> getMappedTasks(TaskMapperContext taskMapperContext) throws TerminateWorkflowException { logger.debug("TaskMapperContext {} in SimpleTaskMapper", taskMapperContext); WorkflowTask taskToSchedule = taskMapperContext.getTaskToSchedule(); Workflow workflowInstance = taskMapperContext.getWorkflowInstance(); int retryCount = taskMapperContext.getRetryCount(); String retriedTaskId = taskMapperContext.getRetryTaskId(); TaskDef taskDefinition = Optional.ofNullable(taskToSchedule.getTaskDefinition()) .orElseThrow(() -> { String reason = String.format("Invalid task. Task %s does not have a definition", taskToSchedule.getName()); return new TerminateWorkflowException(reason); }); Map<String, Object> input = parametersUtils.getTaskInput(taskToSchedule.getInputParameters(), workflowInstance, taskDefinition, taskMapperContext.getTaskId()); Task simpleTask = new Task(); simpleTask.setStartDelayInSeconds(taskToSchedule.getStartDelay()); simpleTask.setTaskId(taskMapperContext.getTaskId()); simpleTask.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); simpleTask.setInputData(input); simpleTask.setWorkflowInstanceId(workflowInstance.getWorkflowId()); simpleTask.setWorkflowType(workflowInstance.getWorkflowName()); simpleTask.setStatus(Task.Status.SCHEDULED); simpleTask.setTaskType(taskToSchedule.getName()); simpleTask.setTaskDefName(taskToSchedule.getName()); simpleTask.setCorrelationId(workflowInstance.getCorrelationId()); simpleTask.setScheduledTime(System.currentTimeMillis()); simpleTask.setRetryCount(retryCount); simpleTask.setCallbackAfterSeconds(taskToSchedule.getStartDelay()); simpleTask.setResponseTimeoutSeconds(taskDefinition.getResponseTimeoutSeconds()); simpleTask.setWorkflowTask(taskToSchedule); simpleTask.setRetriedTaskId(retriedTaskId); simpleTask.setWorkflowPriority(workflowInstance.getPriority()); simpleTask.setRateLimitPerFrequency(taskDefinition.getRateLimitPerFrequency()); simpleTask.setRateLimitFrequencyInSeconds(taskDefinition.getRateLimitFrequencyInSeconds()); return Collections.singletonList(simpleTask); } SimpleTaskMapper(ParametersUtils parametersUtils); @Override List<Task> getMappedTasks(TaskMapperContext taskMapperContext); static final Logger logger; }
@Test public void getMappedTasks() throws Exception { WorkflowTask taskToSchedule = new WorkflowTask(); taskToSchedule.setName("simple_task"); taskToSchedule.setTaskDefinition(new TaskDef("simple_task")); String taskId = IDGenerator.generate(); String retriedTaskId = IDGenerator.generate(); WorkflowDef wd = new WorkflowDef(); Workflow w = new Workflow(); w.setWorkflowDefinition(wd); TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder() .withWorkflowDefinition(wd) .withWorkflowInstance(w) .withTaskDefinition(new TaskDef()) .withTaskToSchedule(taskToSchedule) .withTaskInput(new HashMap<>()) .withRetryCount(0) .withRetryTaskId(retriedTaskId) .withTaskId(taskId) .build(); List<Task> mappedTasks = simpleTaskMapper.getMappedTasks(taskMapperContext); assertNotNull(mappedTasks); assertEquals(1, mappedTasks.size()); }
@Test public void getMappedTasksException() throws Exception { WorkflowTask taskToSchedule = new WorkflowTask(); taskToSchedule.setName("simple_task"); String taskId = IDGenerator.generate(); String retriedTaskId = IDGenerator.generate(); WorkflowDef wd = new WorkflowDef(); Workflow w = new Workflow(); w.setWorkflowDefinition(wd); TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder() .withWorkflowDefinition(wd) .withWorkflowInstance(w) .withTaskDefinition(new TaskDef()) .withTaskToSchedule(taskToSchedule) .withTaskInput(new HashMap<>()) .withRetryCount(0) .withRetryTaskId(retriedTaskId) .withTaskId(taskId) .build(); expectedException.expect(TerminateWorkflowException.class); expectedException.expectMessage(String.format("Invalid task. Task %s does not have a definition", taskToSchedule.getName())); simpleTaskMapper.getMappedTasks(taskMapperContext); }