@Test public void testGetSlotStates() throws Exception { WorkflowID id = wf.getID(); ScheduledTime time1_noSlot = new ScheduledTime("2015-03-05T01:00:00Z"); ScheduledTime time2_slot = new ScheduledTime("2015-03-05T02:00:00Z"); ScheduledTime time3_slot = new ScheduledTime("2015-03-05T03:00:00Z"); ScheduledTime time4_noSlot = new ScheduledTime("2015-03-05T04:00:00Z"); ScheduledTime time5_slot = new ScheduledTime("2015-03-05T05:00:00Z"); ScheduledTime time6_noSlot = new ScheduledTime("2015-03-05T06:00:00Z"); ScheduledTime time7_noSlot = new ScheduledTime("2015-03-05T07:00:00Z"); SortedSet<ScheduledTime> scheduledTimes = Sets.newTreeSet(); scheduledTimes.add(time1_noSlot); scheduledTimes.add(time2_slot); scheduledTimes.add(time5_slot); scheduledTimes.add(time6_noSlot); Map<SlotID, SlotState> dbSlotStates1 = Maps.newHashMap(); SlotID slotID1 = new SlotID(id, time2_slot); dbSlotStates1.put(slotID1, new SlotState(slotID1, SlotState.Status.SUCCESS)); SlotID slotID2 = new SlotID(id, time3_slot); dbSlotStates1.put(slotID2,new SlotState(slotID2, SlotState.Status.SUCCESS)); SlotID slotID3 = new SlotID(id, time5_slot); dbSlotStates1.put(slotID3, new SlotState(slotID3, SlotState.Status.SUCCESS)); when(mockedConnection.getSlotStates(id, time1_noSlot, time7_noSlot)).thenReturn(dbSlotStates1); when(schedule.getScheduledTimes(scheduler, time1_noSlot, time7_noSlot)).thenReturn(scheduledTimes); List<SlotState> slotStates = scheduler.getSlotStates(wf, time1_noSlot, time7_noSlot, mockedConnection); Assert.assertEquals(slotStates.size(), 4); Assert.assertEquals(slotStates.get(0).getStatus(), SlotState.Status.WAITING); Assert.assertEquals(slotStates.get(0).getScheduledTime(), time1_noSlot); Assert.assertEquals(slotStates.get(1).getStatus(), SlotState.Status.SUCCESS); Assert.assertEquals(slotStates.get(1).getScheduledTime(), time2_slot); Assert.assertEquals(slotStates.get(2).getStatus(), SlotState.Status.SUCCESS); Assert.assertEquals(slotStates.get(2).getScheduledTime(), time5_slot); Assert.assertEquals(slotStates.get(3).getStatus(), SlotState.Status.WAITING); Assert.assertEquals(slotStates.get(3).getScheduledTime(), time6_noSlot); }
public List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection) throws Exception { SortedSet<ScheduledTime> times = new TreeSet<>(); times.addAll(wf.getSchedule().getScheduledTimes(this, start, end)); Map<SlotID, SlotState> fetchedSlots = connection.getSlotStates(wf.getID(), start, end); return matchScheduledToFetched(wf, times, fetchedSlots); }
Scheduler { public List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection) throws Exception { SortedSet<ScheduledTime> times = new TreeSet<>(); times.addAll(wf.getSchedule().getScheduledTimes(this, start, end)); Map<SlotID, SlotState> fetchedSlots = connection.getSlotStates(wf.getID(), start, end); return matchScheduledToFetched(wf, times, fetchedSlots); } }
Scheduler { public List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection) throws Exception { SortedSet<ScheduledTime> times = new TreeSet<>(); times.addAll(wf.getSchedule().getScheduledTimes(this, start, end)); Map<SlotID, SlotState> fetchedSlots = connection.getSlotStates(wf.getID(), start, end); return matchScheduledToFetched(wf, times, fetchedSlots); } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); }
Scheduler { public List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection) throws Exception { SortedSet<ScheduledTime> times = new TreeSet<>(); times.addAll(wf.getSchedule().getScheduledTimes(this, start, end)); Map<SlotID, SlotState> fetchedSlots = connection.getSlotStates(wf.getID(), start, end); return matchScheduledToFetched(wf, times, fetchedSlots); } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); void step(ScheduledTime current, StateDatabaseConnection connection); void step(ScheduledTime current, Set<WorkflowID> workflowIDs, StateDatabaseConnection connection); List<SlotState> getSlotStatesIncludingMarkedForRerun(Workflow wf, ScheduledTime current, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); ScheduledTime getWorkflowStartTime(Workflow wf, ScheduledTime current); int getSlidingWindowHours(); WorkflowConfiguration getWorkflowConfiguration(); }
Scheduler { public List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection) throws Exception { SortedSet<ScheduledTime> times = new TreeSet<>(); times.addAll(wf.getSchedule().getScheduledTimes(this, start, end)); Map<SlotID, SlotState> fetchedSlots = connection.getSlotStates(wf.getID(), start, end); return matchScheduledToFetched(wf, times, fetchedSlots); } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); void step(ScheduledTime current, StateDatabaseConnection connection); void step(ScheduledTime current, Set<WorkflowID> workflowIDs, StateDatabaseConnection connection); List<SlotState> getSlotStatesIncludingMarkedForRerun(Workflow wf, ScheduledTime current, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); ScheduledTime getWorkflowStartTime(Workflow wf, ScheduledTime current); int getSlidingWindowHours(); WorkflowConfiguration getWorkflowConfiguration();  }