RedisExecutionDAO extends BaseDynoDAO implements ExecutionDAO { @VisibleForTesting void correlateTaskToWorkflowInDS(String taskId, String workflowInstanceId) { String workflowToTaskKey = nsKey(WORKFLOW_TO_TASKS, workflowInstanceId); dynoClient.sadd(workflowToTaskKey, taskId); logger.debug("Task mapped in WORKFLOW_TO_TASKS with workflowToTaskKey: {}, workflowId: {}, taskId: {}", workflowToTaskKey, workflowInstanceId, taskId); } @Inject  RedisExecutionDAO(DynoProxy dynoClient, ObjectMapper objectMapper, Configuration config); @Override List<Task> getPendingTasksByWorkflow(String taskName, String workflowId); @Override List<Task> getTasks(String taskDefName, String startKey, int count); @Override List<Task> createTasks(List<Task> tasks); @Override void updateTask(Task task); @Override boolean exceedsInProgressLimit(Task task); @Override boolean removeTask(String taskId); @Override Task getTask(String taskId); @Override List<Task> getTasks(List<String> taskIds); @Override List<Task> getTasksForWorkflow(String workflowId); @Override List<Task> getPendingTasksForTaskType(String taskName); @Override String createWorkflow(Workflow workflow); @Override String updateWorkflow(Workflow workflow); @Override boolean removeWorkflow(String workflowId); boolean removeWorkflowWithExpiry(String workflowId, int ttlSeconds); @Override void removeFromPendingWorkflow(String workflowType, String workflowId); @Override Workflow getWorkflow(String workflowId); @Override Workflow getWorkflow(String workflowId, boolean includeTasks); @Override List<String> getRunningWorkflowIds(String workflowName, int version); @Override List<Workflow> getPendingWorkflowsByType(String workflowName, int version); @Override List<Workflow> getWorkflowsByType(String workflowName, Long startTime, Long endTime); @Override List<Workflow> getWorkflowsByCorrelationId(String workflowName, String correlationId, boolean includeTasks); @Override boolean canSearchAcrossWorkflows(); long getPendingWorkflowCount(String workflowName); @Override long getInProgressTaskCount(String taskDefName); @Override boolean addEventExecution(EventExecution eventExecution); @Override void updateEventExecution(EventExecution eventExecution); @Override void removeEventExecution(EventExecution eventExecution); List<EventExecution> getEventExecutions(String eventHandlerName, String eventName, String messageId, int max); static final Logger logger; }
@Test public void testCorrelateTaskToWorkflowInDS() { String workflowId = "workflowId"; String taskId = "taskId1"; String taskDefName = "task1"; TaskDef def = new TaskDef(); def.setName("task1"); def.setConcurrentExecLimit(1); Task task = new Task(); task.setTaskId(taskId); task.setWorkflowInstanceId(workflowId); task.setReferenceTaskName("ref_name"); task.setTaskDefName(taskDefName); task.setTaskType(taskDefName); task.setStatus(Status.IN_PROGRESS); List<Task> tasks = executionDAO.createTasks(Collections.singletonList(task)); assertNotNull(tasks); assertEquals(1, tasks.size()); executionDAO.correlateTaskToWorkflowInDS(taskId, workflowId); tasks = executionDAO.getTasksForWorkflow(workflowId); assertNotNull(tasks); assertEquals(workflowId, tasks.get(0).getWorkflowInstanceId()); assertEquals(taskId, tasks.get(0).getTaskId()); }