@Test public void testGetAndNotFounds() throws Exception { factory.begin(() -> { Project srcProj1 = Project.of("proj1"); Revision srcRev1 = createRevision("rev1"); WorkflowDefinition srcWf1Rev1 = createWorkflow("wf1"); WorkflowDefinition srcWf2 = createWorkflow("wf2"); Revision srcRev2 = createRevision("rev2"); WorkflowDefinition srcWf1Rev2 = createWorkflow("wf1"); WorkflowDefinition srcWf3 = createWorkflow("wf3"); Instant now = Instant.ofEpochSecond(Instant.now().getEpochSecond()); Instant runTime1 = now.plusSeconds(1); Instant schedTime1 = now.plusSeconds(10); Instant runTime2 = now.plusSeconds(2); Instant schedTime2 = now.plusSeconds(20); final AtomicReference<StoredRevision> revRef = new AtomicReference<>(); final AtomicReference<StoredWorkflowDefinition> wfRefA = new AtomicReference<>(); final AtomicReference<StoredWorkflowDefinition> wfRefB = new AtomicReference<>(); StoredProject proj1 = store.putAndLockProject( srcProj1, (store, stored) -> { ProjectControl lock = new ProjectControl(store, stored); revRef.set(lock.insertRevision(srcRev1)); wfRefA.set(lock.insertWorkflowDefinitionsWithoutSchedules(revRef.get(), ImmutableList.of(srcWf1Rev1)).get(0)); wfRefB.set(lock.insertWorkflowDefinitionsWithoutSchedules(revRef.get(), ImmutableList.of(srcWf2)).get(0)); store.updateSchedules( stored.getId(), ImmutableList.of( Schedule.of( srcWf1Rev1.getName(), wfRefA.get().getId(), runTime1, schedTime1), Schedule.of( srcWf2.getName(), wfRefB.get().getId(), runTime1, schedTime1) ), (oldStatus, newSched) -> { return oldStatus.getNextScheduleTime(); }); return lock.get(); }); StoredWorkflowDefinition wf1Rev1 = wfRefA.get(); StoredWorkflowDefinition wf2 = wfRefB.get(); List<StoredSchedule> schedList1 = schedStore.getSchedules(100, Optional.absent()); assertEquals(2, schedList1.size()); StoredSchedule sched1 = schedList1.get(0); StoredSchedule sched2 = schedList1.get(1); assertThat(schedStore.getSchedulesByProjectId(proj1.getId(), 100, Optional.absent()), is(schedList1)); assertThat(schedStore.getScheduleByProjectIdAndWorkflowName(proj1.getId(), srcWf1Rev1.getName()), is(sched1)); assertThat(schedStore.getScheduleByProjectIdAndWorkflowName(proj1.getId(), srcWf2.getName()), is(sched2)); assertThat(schedStore.getSchedulesByProjectId(4711, 100, Optional.absent()), is(empty())); try { schedStore.getScheduleByProjectIdAndWorkflowName(proj1.getId(), "non-existent-workflow"); fail(); } catch (ResourceNotFoundException ignore) { } try { schedStore.getScheduleByProjectIdAndWorkflowName(4711, srcWf1Rev1.getName()); fail(); } catch (ResourceNotFoundException ignore) { } store.putAndLockProject( srcProj1, (store, stored) -> { ProjectControl lock = new ProjectControl(store, stored); revRef.set(lock.insertRevision(srcRev2)); wfRefA.set(lock.insertWorkflowDefinitionsWithoutSchedules(revRef.get(), ImmutableList.of(srcWf1Rev2)).get(0)); wfRefB.set(lock.insertWorkflowDefinitionsWithoutSchedules(revRef.get(), ImmutableList.of(srcWf3)).get(0)); store.updateSchedules( stored.getId(), ImmutableList.of( Schedule.of( srcWf1Rev2.getName(), wfRefA.get().getId(), runTime1, schedTime2), Schedule.of( srcWf3.getName(), wfRefB.get().getId(), runTime2, schedTime2) ), (oldStatus, newSched) -> { return ScheduleTime.of( oldStatus.getNextScheduleTime().getTime().minusSeconds(60), oldStatus.getNextScheduleTime().getRunTime().minusSeconds(120)); }); return lock.get(); }); StoredWorkflowDefinition wf1Rev2 = wfRefA.get(); StoredWorkflowDefinition wf3 = wfRefB.get(); List<StoredSchedule> schedList2 = schedStore.getSchedules(100, Optional.absent()); assertEquals(2, schedList2.size()); StoredSchedule sched3 = schedList2.get(0); StoredSchedule sched4 = schedList2.get(1); assertEquals(proj1.getId(), sched1.getProjectId()); assertEquals(proj1.getId(), sched2.getProjectId()); assertEquals(wf1Rev1.getName(), sched1.getWorkflowName()); assertEquals(wf2.getName(), sched2.getWorkflowName()); assertEquals(wf1Rev1.getId(), sched1.getWorkflowDefinitionId()); assertEquals(wf2.getId(), sched2.getWorkflowDefinitionId()); assertEquals(proj1.getId(), sched3.getProjectId()); assertEquals(proj1.getId(), sched4.getProjectId()); assertEquals(wf1Rev2.getName(), sched3.getWorkflowName()); assertEquals(wf3.getName(), sched4.getWorkflowName()); assertEquals(wf1Rev2.getId(), sched3.getWorkflowDefinitionId()); assertEquals(wf3.getId(), sched4.getWorkflowDefinitionId()); assertEquals(runTime1, sched1.getNextRunTime()); assertEquals(runTime1, sched2.getNextRunTime()); assertEquals(schedTime1, sched1.getNextScheduleTime()); assertEquals(schedTime1, sched2.getNextScheduleTime()); assertEquals(runTime1.minusSeconds(120), sched3.getNextRunTime()); assertEquals(runTime2, sched4.getNextRunTime()); assertEquals(schedTime1.minusSeconds(60), sched3.getNextScheduleTime()); assertEquals(schedTime2, sched4.getNextScheduleTime()); assertEquals(sched3.getId(), sched1.getId()); assertNotFound(() -> schedStore.getScheduleById(sched2.getId())); assertEquals(sched3, schedStore.getScheduleById(sched3.getId())); assertEquals(sched4, schedStore.getScheduleById(sched4.getId())); assertEquals(sched4.getId(), (long) schedStore.lockScheduleById(sched4.getId(), (store, schedule) -> schedule.getId())); List<Integer> lockedByRuntime1 = new ArrayList<>(); schedManager.lockReadySchedules(runTime1, 10, (store, schedule) -> { lockedByRuntime1.add(schedule.getId()); }); assertEquals(ImmutableList.of(sched1.getId()), lockedByRuntime1); List<Integer> lockedByRuntime2 = new ArrayList<>(); schedManager.lockReadySchedules(runTime2, 10, (store, schedule) -> { lockedByRuntime2.add(schedule.getId()); }); assertEquals(ImmutableList.of(sched3.getId(), sched4.getId()), lockedByRuntime2); try { schedManager.lockReadySchedules(runTime2, 10, (store, schedule) -> { throw new RuntimeException("processing " + schedule.getId()); }); fail(); } catch (RuntimeException ex) { assertEquals(ex.getMessage(), "processing " + sched3.getId()); assertEquals(1, ex.getSuppressed().length); assertEquals(ex.getSuppressed()[0].getMessage(), "processing " + sched4.getId()); } Instant runTime3 = now.plusSeconds(3); Instant schedTime3 = now.plusSeconds(30); Instant runTime4 = now.plusSeconds(4); Instant schedTime4 = now.plusSeconds(40); try { schedManager.lockReadySchedules(runTime2, 10, (store, schedule) -> { if (schedule.getId() == sched3.getId()) { throw new RuntimeException(); } else { try { store.updateNextScheduleTime(schedule.getId(), ScheduleTime.of(schedTime3, runTime3)); } catch (ResourceNotFoundException ex) { throw new RuntimeException(ex); } } }); fail(); } catch (RuntimeException ex) { } List<Integer> updated = new ArrayList<>(); schedManager.lockReadySchedules(runTime2, 10, (store, schedule) -> { updated.add(schedule.getId()); try { store.updateNextScheduleTimeAndLastSessionTime(schedule.getId(), ScheduleTime.of(schedTime4, runTime4), schedTime1); } catch (ResourceNotFoundException ex) { throw new RuntimeException(ex); } }); assertEquals(ImmutableList.of(sched1.getId()), updated); StoredSchedule updatedSched1 = schedStore.getSchedules(100, Optional.absent()).get(0); assertEquals(sched1.getId(), updatedSched1.getId()); assertEquals(runTime4, updatedSched1.getNextRunTime()); assertEquals(schedTime4, updatedSched1.getNextScheduleTime()); }); }
@Override public int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func) { List<RuntimeException> exceptions = new ArrayList<>(); long count = transaction((handle, dao) -> { Stream<StoredSchedule> schedStream; if (dao instanceof PgDao) { schedStream = ((PgDao) dao).lockReadySchedulesSkipLocked(currentTime.getEpochSecond(), limit) .stream(); } else { schedStream = dao.lockReadyScheduleIds(currentTime.getEpochSecond(), limit) .stream() .map(dao::getScheduleByIdInternal); } return schedStream.mapToInt(sched -> { try { func.schedule(new DatabaseScheduleControlStore(handle), sched); } catch (RuntimeException ex) { exceptions.add(ex); } return 1; }) .sum(); }); if (!exceptions.isEmpty()) { RuntimeException first = exceptions.get(0); for (RuntimeException ex : exceptions.subList(1, exceptions.size())) { first.addSuppressed(ex); } throw first; } return (int) count; }
DatabaseScheduleStoreManager extends BasicDatabaseStoreManager<DatabaseScheduleStoreManager.Dao> implements ScheduleStoreManager { @Override public int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func) { List<RuntimeException> exceptions = new ArrayList<>(); long count = transaction((handle, dao) -> { Stream<StoredSchedule> schedStream; if (dao instanceof PgDao) { schedStream = ((PgDao) dao).lockReadySchedulesSkipLocked(currentTime.getEpochSecond(), limit) .stream(); } else { schedStream = dao.lockReadyScheduleIds(currentTime.getEpochSecond(), limit) .stream() .map(dao::getScheduleByIdInternal); } return schedStream.mapToInt(sched -> { try { func.schedule(new DatabaseScheduleControlStore(handle), sched); } catch (RuntimeException ex) { exceptions.add(ex); } return 1; }) .sum(); }); if (!exceptions.isEmpty()) { RuntimeException first = exceptions.get(0); for (RuntimeException ex : exceptions.subList(1, exceptions.size())) { first.addSuppressed(ex); } throw first; } return (int) count; } }
DatabaseScheduleStoreManager extends BasicDatabaseStoreManager<DatabaseScheduleStoreManager.Dao> implements ScheduleStoreManager { @Override public int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func) { List<RuntimeException> exceptions = new ArrayList<>(); long count = transaction((handle, dao) -> { Stream<StoredSchedule> schedStream; if (dao instanceof PgDao) { schedStream = ((PgDao) dao).lockReadySchedulesSkipLocked(currentTime.getEpochSecond(), limit) .stream(); } else { schedStream = dao.lockReadyScheduleIds(currentTime.getEpochSecond(), limit) .stream() .map(dao::getScheduleByIdInternal); } return schedStream.mapToInt(sched -> { try { func.schedule(new DatabaseScheduleControlStore(handle), sched); } catch (RuntimeException ex) { exceptions.add(ex); } return 1; }) .sum(); }); if (!exceptions.isEmpty()) { RuntimeException first = exceptions.get(0); for (RuntimeException ex : exceptions.subList(1, exceptions.size())) { first.addSuppressed(ex); } throw first; } return (int) count; } @Inject  DatabaseScheduleStoreManager(TransactionManager transactionManager, ConfigMapper cfm, DatabaseConfig config); }
DatabaseScheduleStoreManager extends BasicDatabaseStoreManager<DatabaseScheduleStoreManager.Dao> implements ScheduleStoreManager { @Override public int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func) { List<RuntimeException> exceptions = new ArrayList<>(); long count = transaction((handle, dao) -> { Stream<StoredSchedule> schedStream; if (dao instanceof PgDao) { schedStream = ((PgDao) dao).lockReadySchedulesSkipLocked(currentTime.getEpochSecond(), limit) .stream(); } else { schedStream = dao.lockReadyScheduleIds(currentTime.getEpochSecond(), limit) .stream() .map(dao::getScheduleByIdInternal); } return schedStream.mapToInt(sched -> { try { func.schedule(new DatabaseScheduleControlStore(handle), sched); } catch (RuntimeException ex) { exceptions.add(ex); } return 1; }) .sum(); }); if (!exceptions.isEmpty()) { RuntimeException first = exceptions.get(0); for (RuntimeException ex : exceptions.subList(1, exceptions.size())) { first.addSuppressed(ex); } throw first; } return (int) count; } @Inject  DatabaseScheduleStoreManager(TransactionManager transactionManager, ConfigMapper cfm, DatabaseConfig config); @Override ScheduleStore getScheduleStore(int siteId); @Override int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func); }
DatabaseScheduleStoreManager extends BasicDatabaseStoreManager<DatabaseScheduleStoreManager.Dao> implements ScheduleStoreManager { @Override public int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func) { List<RuntimeException> exceptions = new ArrayList<>(); long count = transaction((handle, dao) -> { Stream<StoredSchedule> schedStream; if (dao instanceof PgDao) { schedStream = ((PgDao) dao).lockReadySchedulesSkipLocked(currentTime.getEpochSecond(), limit) .stream(); } else { schedStream = dao.lockReadyScheduleIds(currentTime.getEpochSecond(), limit) .stream() .map(dao::getScheduleByIdInternal); } return schedStream.mapToInt(sched -> { try { func.schedule(new DatabaseScheduleControlStore(handle), sched); } catch (RuntimeException ex) { exceptions.add(ex); } return 1; }) .sum(); }); if (!exceptions.isEmpty()) { RuntimeException first = exceptions.get(0); for (RuntimeException ex : exceptions.subList(1, exceptions.size())) { first.addSuppressed(ex); } throw first; } return (int) count; } @Inject  DatabaseScheduleStoreManager(TransactionManager transactionManager, ConfigMapper cfm, DatabaseConfig config); @Override ScheduleStore getScheduleStore(int siteId); @Override int lockReadySchedules(Instant currentTime, int limit, ScheduleAction func);  }