Workflow extends AbstractWorkflowItem implements WorkflowStatusListener { public void add(WorkflowItem... items) { for (WorkflowItem item : items) { item.addWorkflowStatusListener(this); } itemList.addAll(Arrays.asList(items)); if (!itemList.isEmpty()) { aggregateChildInformation(); } } protected  Workflow(WorkflowItem... items); void add(WorkflowItem... items); boolean isSustainable(); void setSustainable(boolean sustainable); @Override void kill(); @Override void updateStatus(); @Override void handleStatusChanged(WorkflowStatusEvent event); @Override Object[] getJobIds(); @Override WorkflowItem[] getItems();  }
@Test public void testSequential() throws Exception { Workflow.Sequential wf = new Workflow.Sequential(); LifeStepWorkflowItem job1 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job2 = new LifeStepWorkflowItem(); wf.add(job1, job2); wf.submit(); assertEquals(ProcessState.UNKNOWN, job1.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job2.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.SCHEDULED, job1.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job2.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.RUNNING, job1.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job2.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, job2.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.RUNNING, job2.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wf.getStatus().getState()); assertEquals(1, job1.getSubmitCount()); assertEquals(1, job2.getSubmitCount()); }
@Test public void testParallel() throws Exception { Workflow.Parallel wf = new Workflow.Parallel(); LifeStepWorkflowItem job1 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job2 = new LifeStepWorkflowItem(); wf.add(job1, job2); wf.submit(); assertEquals(ProcessState.UNKNOWN, job1.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job2.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.SCHEDULED, job1.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, job2.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.RUNNING, job1.getStatus().getState()); assertEquals(ProcessState.RUNNING, job2.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wf.getStatus().getState()); incLifeStep(job1, job2); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wf.getStatus().getState()); assertEquals(1, job1.getSubmitCount()); assertEquals(1, job2.getSubmitCount()); }
@Test public void testMixed() throws Exception { Workflow.Sequential wf = new Workflow.Sequential(); LifeStepWorkflowItem job1 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job2 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job3 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job4 = new LifeStepWorkflowItem(); Workflow.Parallel wfp = new Workflow.Parallel(); wfp.add(job1, job2); Workflow.Sequential wfs = new Workflow.Sequential(); wfs.add(job3, job4); wf.add(wfp, wfs); wf.submit(); assertEquals(ProcessState.UNKNOWN, job1.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job2.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job3.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job4.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wfp.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wfs.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.SCHEDULED, job1.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, job2.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job3.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job4.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, wfp.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wfs.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.RUNNING, job1.getStatus().getState()); assertEquals(ProcessState.RUNNING, job2.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job3.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job4.getStatus().getState()); assertEquals(ProcessState.RUNNING, wfp.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, wfs.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, job3.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job4.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfp.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, wfs.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.RUNNING, job3.getStatus().getState()); assertEquals(ProcessState.UNKNOWN, job4.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfp.getStatus().getState()); assertEquals(ProcessState.RUNNING, wfs.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job3.getStatus().getState()); assertEquals(ProcessState.SCHEDULED, job4.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfp.getStatus().getState()); assertEquals(ProcessState.RUNNING, wfs.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job3.getStatus().getState()); assertEquals(ProcessState.RUNNING, job4.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfp.getStatus().getState()); assertEquals(ProcessState.RUNNING, wfs.getStatus().getState()); assertEquals(ProcessState.RUNNING, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job3.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job4.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfp.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfs.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wf.getStatus().getState()); incLifeStep(job1, job2, job3, job4); assertEquals(ProcessState.COMPLETED, job1.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job2.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job3.getStatus().getState()); assertEquals(ProcessState.COMPLETED, job4.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfp.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wfs.getStatus().getState()); assertEquals(ProcessState.COMPLETED, wf.getStatus().getState()); assertEquals(1, job1.getSubmitCount()); assertEquals(1, job2.getSubmitCount()); assertEquals(1, job3.getSubmitCount()); assertEquals(1, job4.getSubmitCount()); }
@Test public void testParallelStatusChangeAggregation() { LifeStepWorkflowItem job1 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job2 = new LifeStepWorkflowItem(); Workflow wf = new Workflow.Parallel(); wf.add(job1, job2); job1.setStatus(new ProcessStatus(ProcessState.RUNNING, 0.4F)); job2.setStatus(new ProcessStatus(ProcessState.RUNNING, 0.2F)); assertEquals(new ProcessStatus(ProcessState.RUNNING, 0.3F), wf.getStatus()); job2.setStatus(new ProcessStatus(ProcessState.RUNNING, 0.8F)); assertEquals(new ProcessStatus(ProcessState.RUNNING, 0.6F), wf.getStatus()); job1.setStatus(new ProcessStatus(ProcessState.RUNNING, 0.8F)); assertEquals(new ProcessStatus(ProcessState.RUNNING, 0.8F), wf.getStatus()); }
@Test public void testSequentialStatusChangeAggregation() { LifeStepWorkflowItem job1 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job2 = new LifeStepWorkflowItem(); Workflow wf = new Workflow.Sequential(); wf.add(job1, job2); job2.setStatus(new ProcessStatus(ProcessState.RUNNING, 0.3F)); job1.setStatus(new ProcessStatus(ProcessState.UNKNOWN)); assertEquals(new ProcessStatus(ProcessState.RUNNING, 0.15F), wf.getStatus()); }
@Test public void testTimeTrackingForSequentialWorkflow() throws WorkflowException { Workflow.Sequential wf = new Workflow.Sequential(); LifeStepWorkflowItem job1 = new LifeStepWorkflowItem(); LifeStepWorkflowItem job2 = new LifeStepWorkflowItem(); wf.add(job1, job2); wf.submit(); Date submitTime1 = new Date(2000); Date startTime1 = new Date(3000); Date stopTime1 = new Date(5000); Date submitTime2 = new Date(5000); Date startTime2 = new Date(7000); Date stopTime2 = new Date(9000); assertWfTime(job1, null, null, null); assertWfTime(job2, null, null, null); assertWfTime(wf, null, null, null); setTime(submitTime1, job1, job2); incLifeStep(job1, job2); assertWfTime(job1, submitTime1, null, null); assertWfTime(job2, null, null, null); assertWfTime(wf, submitTime1, null, null); setTime(startTime1, job1, job2); incLifeStep(job1, job2); assertWfTime(job1, submitTime1, startTime1, null); assertWfTime(job2, null, null, null); assertWfTime(wf, submitTime1, startTime1, null); setTime(stopTime1, job1, job2); incLifeStep(job1, job2); assertWfTime(job1, submitTime1, startTime1, stopTime1); assertWfTime(job2, submitTime2, null, null); assertWfTime(wf, submitTime1, startTime1, null); setTime(startTime2, job1, job2); incLifeStep(job1, job2); assertWfTime(job1, submitTime1, startTime1, stopTime1); assertWfTime(job2, submitTime2, startTime2, null); assertWfTime(wf, submitTime1, startTime1, null); setTime(stopTime2, job1, job2); incLifeStep(job1, job2); assertWfTime(job1, submitTime1, startTime1, stopTime1); assertWfTime(job2, submitTime2, startTime2, stopTime2); assertWfTime(wf, submitTime1, startTime1, stopTime2); incLifeStep(job1, job2); assertWfTime(job1, submitTime1, startTime1, stopTime1); assertWfTime(job2, submitTime2, startTime2, stopTime2); assertWfTime(wf, submitTime1, startTime1, stopTime2); }