@Test public void testDisableEnable() throws Exception { factory.begin(() -> { Project srcProj1 = Project.of("proj1"); Revision srcRev1 = createRevision("rev1"); WorkflowDefinition srcWf1Rev1 = createWorkflow("wf1"); WorkflowDefinition srcWf2 = createWorkflow("wf2"); Instant yesterday = Instant.now().minus(Duration.ofDays(1)).truncatedTo(SECONDS); 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(), yesterday, yesterday), Schedule.of( srcWf2.getName(), wfRefB.get().getId(), yesterday, yesterday) ), (oldStatus, newSched) -> { return oldStatus.getNextScheduleTime(); }); return lock.get(); }); StoredWorkflowDefinition wf1 = 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); schedStore.updateScheduleById(sched1.getId(), (store, schedule) -> { store.enableSchedule(schedule.getId()); return schedule; }); { List<Integer> ready = new ArrayList<>(); schedManager.lockReadySchedules(Instant.now(), 10, (store, schedule) -> ready.add(schedule.getId())); assertThat(ready, containsInAnyOrder(sched1.getId(), sched2.getId())); } schedStore.updateScheduleById(sched1.getId(), (store, schedule) -> { store.disableSchedule(schedule.getId()); return schedule; }); { List<Integer> ready = new ArrayList<>(); schedManager.lockReadySchedules(Instant.now(), 10, (store, schedule) -> ready.add(schedule.getId())); assertThat(ready, contains(sched2.getId())); } { double now = Instant.now().getEpochSecond(); StoredSchedule s1 = schedStore.getScheduleByProjectIdAndWorkflowName(proj1.getId(), wf1.getName()); StoredSchedule s2 = schedStore.getScheduleByProjectIdAndWorkflowName(proj1.getId(), wf2.getName()); assertThat(s1.getId(), is(sched1.getId())); assertThat(s2.getId(), is(sched2.getId())); assertThat((double) s1.getDisabledAt().get().getEpochSecond(), is(closeTo(now, 30))); assertThat(s2.getDisabledAt(), is(Optional.absent())); assertThat(schedStore.getSchedules(100, Optional.absent()), containsInAnyOrder(s1, s2)); assertThat(schedStore.getSchedulesByProjectId(proj1.getId(), 100, Optional.absent()), containsInAnyOrder(s1, s2)); } schedStore.updateScheduleById(sched1.getId(), (store, schedule) -> { store.enableSchedule(schedule.getId()); return schedule; }); { List<Integer> ready = new ArrayList<>(); schedManager.lockReadySchedules(Instant.now(), 10, (store, schedule) -> ready.add(schedule.getId())); assertThat(ready, containsInAnyOrder(sched1.getId(), sched2.getId())); } schedStore.updateScheduleById(sched1.getId(), (store, schedule) -> { store.enableSchedule(schedule.getId()); return schedule; }); { List<Integer> ready = new ArrayList<>(); schedManager.lockReadySchedules(Instant.now(), 10, (store, schedule) -> ready.add(schedule.getId())); assertThat(ready, containsInAnyOrder(sched1.getId(), sched2.getId())); } }); }
@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);  }