DeciderService { public DeciderOutcome decide(Workflow workflow) throws TerminateWorkflowException { final List<Task> tasks = workflow.getTasks(); List<Task> unprocessedTasks = tasks.stream() .filter(t -> !t.getStatus().equals(SKIPPED) && !t.isExecuted()) .collect(Collectors.toList()); List<Task> tasksToBeScheduled = new LinkedList<>(); if (unprocessedTasks.isEmpty()) { tasksToBeScheduled = startWorkflow(workflow); if (tasksToBeScheduled == null) { tasksToBeScheduled = new LinkedList<>(); } } return decide(workflow, tasksToBeScheduled); } @Inject  DeciderService(ParametersUtils parametersUtils, MetadataDAO metadataDAO,
                          ExternalPayloadStorageUtils externalPayloadStorageUtils,
                          @Named("TaskMappers") Map<String, TaskMapper> taskMappers, Configuration configuration); DeciderOutcome decide(Workflow workflow); List<Task> getTasksToBeScheduled(Workflow workflow,
                                            WorkflowTask taskToSchedule, int retryCount); List<Task> getTasksToBeScheduled(Workflow workflow,
                                            WorkflowTask taskToSchedule, int retryCount, String retriedTaskId);  }
@Test public void testCaseStatement() { WorkflowDef def = createConditionalWF(); Workflow wf = new Workflow(); wf.setWorkflowDefinition(def); wf.setCreateTime(0L); wf.setWorkflowId("a"); wf.setCorrelationId("b"); wf.setStatus(WorkflowStatus.RUNNING); DeciderOutcome outcome = deciderService.decide(wf); List<Task> scheduledTasks = outcome.tasksToBeScheduled; assertNotNull(scheduledTasks); assertEquals(2, scheduledTasks.size()); assertEquals(Status.IN_PROGRESS, scheduledTasks.get(0).getStatus()); assertEquals(Status.SCHEDULED, scheduledTasks.get(1).getStatus()); }
@Test public void testFork() throws IOException { InputStream stream = TestDeciderService.class.getResourceAsStream("/test.json"); Workflow workflow = objectMapper.readValue(stream, Workflow.class); DeciderOutcome outcome = deciderService.decide(workflow); assertFalse(outcome.isComplete); assertEquals(5, outcome.tasksToBeScheduled.size()); assertEquals(1, outcome.tasksToBeUpdated.size()); }
@Test public void testDecideSuccessfulWorkflow() { WorkflowDef workflowDef = createLinearWorkflow(); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(workflowDef); workflow.setStatus(WorkflowStatus.RUNNING); Task task1 = new Task(); task1.setTaskType("junit_task_l1"); task1.setReferenceTaskName("s1"); task1.setSeq(1); task1.setRetried(false); task1.setExecuted(false); task1.setStatus(Status.COMPLETED); workflow.getTasks().add(task1); DeciderOutcome deciderOutcome = deciderService.decide(workflow); assertNotNull(deciderOutcome); assertFalse(workflow.getTaskByRefName("s1").isRetried()); assertEquals(1, deciderOutcome.tasksToBeUpdated.size()); assertEquals("s1", deciderOutcome.tasksToBeUpdated.get(0).getReferenceTaskName()); assertEquals(1, deciderOutcome.tasksToBeScheduled.size()); assertEquals("s2", deciderOutcome.tasksToBeScheduled.get(0).getReferenceTaskName()); assertFalse(deciderOutcome.isComplete); Task task2 = new Task(); task2.setTaskType("junit_task_l2"); task2.setReferenceTaskName("s2"); task2.setSeq(2); task2.setRetried(false); task2.setExecuted(false); task2.setStatus(Status.COMPLETED); workflow.getTasks().add(task2); deciderOutcome = deciderService.decide(workflow); assertNotNull(deciderOutcome); assertTrue(workflow.getTaskByRefName("s2").isExecuted()); assertFalse(workflow.getTaskByRefName("s2").isRetried()); assertEquals(1, deciderOutcome.tasksToBeUpdated.size()); assertEquals("s2", deciderOutcome.tasksToBeUpdated.get(0).getReferenceTaskName()); assertEquals(0, deciderOutcome.tasksToBeScheduled.size()); assertTrue(deciderOutcome.isComplete); }
@Test public void testDecideWithLoopTask() { WorkflowDef workflowDef = createLinearWorkflow(); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(workflowDef); workflow.setStatus(WorkflowStatus.RUNNING); Task task1 = new Task(); task1.setTaskType("junit_task_l1"); task1.setReferenceTaskName("s1"); task1.setSeq(1); task1.setIteration(1); task1.setRetried(false); task1.setExecuted(false); task1.setStatus(Status.COMPLETED); workflow.getTasks().add(task1); DeciderOutcome deciderOutcome = deciderService.decide(workflow); assertNotNull(deciderOutcome); assertFalse(workflow.getTaskByRefName("s1").isRetried()); assertEquals(1, deciderOutcome.tasksToBeUpdated.size()); assertEquals("s1", deciderOutcome.tasksToBeUpdated.get(0).getReferenceTaskName()); assertEquals(1, deciderOutcome.tasksToBeScheduled.size()); assertEquals("s2__1", deciderOutcome.tasksToBeScheduled.get(0).getReferenceTaskName()); assertFalse(deciderOutcome.isComplete); }
@Test public void testDecideFailedTask() { WorkflowDef workflowDef = createLinearWorkflow(); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(workflowDef); workflow.setStatus(WorkflowStatus.RUNNING); Task task = new Task(); task.setTaskType("junit_task_l1"); task.setReferenceTaskName("s1"); task.setSeq(1); task.setRetried(false); task.setExecuted(false); task.setStatus(Status.FAILED); WorkflowTask workflowTask = new WorkflowTask(); workflowTask.setTaskReferenceName("s1"); workflowTask.setName("junit_task_l1"); workflowTask.setTaskDefinition(new TaskDef("junit_task_l1")); task.setWorkflowTask(workflowTask); workflow.getTasks().add(task); DeciderOutcome deciderOutcome = deciderService.decide(workflow); assertNotNull(deciderOutcome); assertFalse(workflow.getTaskByRefName("s1").isExecuted()); assertTrue(workflow.getTaskByRefName("s1").isRetried()); assertEquals(1, deciderOutcome.tasksToBeUpdated.size()); assertEquals("s1", deciderOutcome.tasksToBeUpdated.get(0).getReferenceTaskName()); assertEquals(1, deciderOutcome.tasksToBeScheduled.size()); assertEquals("s1", deciderOutcome.tasksToBeScheduled.get(0).getReferenceTaskName()); assertFalse(deciderOutcome.isComplete); }