@Test public void runExternalWorkflowsCallsSchedulerCorrectly() throws Exception { List<SlotState> slotStates = candidate(SlotState.Status.READY); when(externalService.submit(slotId)).thenReturn("externalId"); scheduler.runExternalWorkflows(wf, slotStates, mockedConnection); InOrder inOrder = inOrder(externalService); inOrder.verify(externalService).submit(slotStates.get(0).getSlotID()); inOrder.verify(externalService).start(slotId, "externalId"); verifyNoMoreInteractions(externalService); }
void runExternalWorkflows(Workflow wf, List<SlotState> slotStates, StateDatabaseConnection connection) throws Exception { List<SlotState> scheduledSlots = wf.getSchedulingStrategy().getSchedulingCandidates(slotStates); for (SlotState slotState : scheduledSlots) { if (!slotState.getStatus().equals(SlotState.Status.READY)) { throw new IllegalStateException("Scheduling strategy returned non-ready slot: " + slotState); } SlotID slotID = slotState.getSlotID(); LOGGER.info("Submitting slot to external service: " + slotID); String externalID = wf.getExternalService().submit(slotID); connection.putSlotState(slotState.transitionToRunning(externalID)); LOGGER.info("Starting slot: " + slotID + " with external ID: " + externalID); wf.getExternalService().start(slotID, externalID); } }
Scheduler { void runExternalWorkflows(Workflow wf, List<SlotState> slotStates, StateDatabaseConnection connection) throws Exception { List<SlotState> scheduledSlots = wf.getSchedulingStrategy().getSchedulingCandidates(slotStates); for (SlotState slotState : scheduledSlots) { if (!slotState.getStatus().equals(SlotState.Status.READY)) { throw new IllegalStateException("Scheduling strategy returned non-ready slot: " + slotState); } SlotID slotID = slotState.getSlotID(); LOGGER.info("Submitting slot to external service: " + slotID); String externalID = wf.getExternalService().submit(slotID); connection.putSlotState(slotState.transitionToRunning(externalID)); LOGGER.info("Starting slot: " + slotID + " with external ID: " + externalID); wf.getExternalService().start(slotID, externalID); } } }
Scheduler { void runExternalWorkflows(Workflow wf, List<SlotState> slotStates, StateDatabaseConnection connection) throws Exception { List<SlotState> scheduledSlots = wf.getSchedulingStrategy().getSchedulingCandidates(slotStates); for (SlotState slotState : scheduledSlots) { if (!slotState.getStatus().equals(SlotState.Status.READY)) { throw new IllegalStateException("Scheduling strategy returned non-ready slot: " + slotState); } SlotID slotID = slotState.getSlotID(); LOGGER.info("Submitting slot to external service: " + slotID); String externalID = wf.getExternalService().submit(slotID); connection.putSlotState(slotState.transitionToRunning(externalID)); LOGGER.info("Starting slot: " + slotID + " with external ID: " + externalID); wf.getExternalService().start(slotID, externalID); } } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); }
Scheduler { void runExternalWorkflows(Workflow wf, List<SlotState> slotStates, StateDatabaseConnection connection) throws Exception { List<SlotState> scheduledSlots = wf.getSchedulingStrategy().getSchedulingCandidates(slotStates); for (SlotState slotState : scheduledSlots) { if (!slotState.getStatus().equals(SlotState.Status.READY)) { throw new IllegalStateException("Scheduling strategy returned non-ready slot: " + slotState); } SlotID slotID = slotState.getSlotID(); LOGGER.info("Submitting slot to external service: " + slotID); String externalID = wf.getExternalService().submit(slotID); connection.putSlotState(slotState.transitionToRunning(externalID)); LOGGER.info("Starting slot: " + slotID + " with external ID: " + externalID); wf.getExternalService().start(slotID, externalID); } } 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 { void runExternalWorkflows(Workflow wf, List<SlotState> slotStates, StateDatabaseConnection connection) throws Exception { List<SlotState> scheduledSlots = wf.getSchedulingStrategy().getSchedulingCandidates(slotStates); for (SlotState slotState : scheduledSlots) { if (!slotState.getStatus().equals(SlotState.Status.READY)) { throw new IllegalStateException("Scheduling strategy returned non-ready slot: " + slotState); } SlotID slotID = slotState.getSlotID(); LOGGER.info("Submitting slot to external service: " + slotID); String externalID = wf.getExternalService().submit(slotID); connection.putSlotState(slotState.transitionToRunning(externalID)); LOGGER.info("Starting slot: " + slotID + " with external ID: " + externalID); wf.getExternalService().start(slotID, externalID); } } 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();  }