@Test void canStartLeadershipTasks() { final long task1Period = 13238; Mockito.when(this.task1.getScheduleType()).thenReturn(GenieTaskScheduleType.FIXED_RATE); Mockito.when(this.task1.getFixedRate()).thenReturn(task1Period); final long task2Period = 3891082; Mockito.when(this.task2.getScheduleType()).thenReturn(GenieTaskScheduleType.FIXED_DELAY); Mockito.when(this.task2.getFixedDelay()).thenReturn(task2Period); final Trigger task3Trigger = Mockito.mock(Trigger.class); Mockito.when(this.task3.getScheduleType()).thenReturn(GenieTaskScheduleType.TRIGGER); Mockito.when(this.task3.getTrigger()).thenReturn(task3Trigger); final OnGrantedEvent event = new OnGrantedEvent(this, null, "blah"); this.coordinator.onLeaderEvent(event); Mockito.verify(this.scheduler, Mockito.times(1)).scheduleAtFixedRate(this.task1, task1Period); Mockito.verify(this.task1, Mockito.never()).getFixedDelay(); Mockito.verify(this.task1, Mockito.never()).getTrigger(); Mockito.verify(this.scheduler, Mockito.times(1)).scheduleWithFixedDelay(this.task2, task2Period); Mockito.verify(this.task2, Mockito.never()).getFixedRate(); Mockito.verify(this.task2, Mockito.never()).getTrigger(); Mockito.verify(this.scheduler, Mockito.times(1)).schedule(this.task3, task3Trigger); Mockito.verify(this.task3, Mockito.never()).getFixedRate(); Mockito.verify(this.task3, Mockito.never()).getFixedDelay(); this.coordinator.onLeaderEvent(event); Mockito.verify(this.scheduler, Mockito.times(1)).scheduleAtFixedRate(this.task1, task1Period); Mockito.verify(this.scheduler, Mockito.times(1)).scheduleWithFixedDelay(this.task2, task2Period); Mockito.verify(this.scheduler, Mockito.times(1)).schedule(this.task3, task3Trigger); }
@EventListener public synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent) { if (leaderEvent instanceof OnGrantedEvent) { if (this.isRunning) { return; } log.info("Leadership granted."); this.isRunning = true; this.tasks.forEach( task -> { switch (task.getScheduleType()) { case TRIGGER: final Trigger trigger = task.getTrigger(); log.info( "Scheduling leadership task {} to run with trigger {}", task.getClass().getCanonicalName(), trigger ); this.futures.add(this.taskScheduler.schedule(task, trigger)); break; case FIXED_RATE: final long rate = task.getFixedRate(); log.info( "Scheduling leadership task {} to run every {} second(s)", task.getClass().getCanonicalName(), rate / 1000.0 ); this.futures.add(this.taskScheduler.scheduleAtFixedRate(task, rate)); break; case FIXED_DELAY: final long delay = task.getFixedDelay(); log.info( "Scheduling leadership task {} to run at a fixed delay of every {} second(s)", task.getClass().getCanonicalName(), delay / 1000.0 ); this.futures.add(this.taskScheduler.scheduleWithFixedDelay(task, delay)); break; default: log.error("Unknown Genie task type {}", task.getScheduleType()); } } ); } else if (leaderEvent instanceof OnRevokedEvent) { if (!this.isRunning) { return; } log.info("Leadership revoked."); this.isRunning = false; this.cancelTasks(); } else { log.warn("Unknown leadership event {}. Ignoring.", leaderEvent); } }
LeaderTasksCoordinator { @EventListener public synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent) { if (leaderEvent instanceof OnGrantedEvent) { if (this.isRunning) { return; } log.info("Leadership granted."); this.isRunning = true; this.tasks.forEach( task -> { switch (task.getScheduleType()) { case TRIGGER: final Trigger trigger = task.getTrigger(); log.info( "Scheduling leadership task {} to run with trigger {}", task.getClass().getCanonicalName(), trigger ); this.futures.add(this.taskScheduler.schedule(task, trigger)); break; case FIXED_RATE: final long rate = task.getFixedRate(); log.info( "Scheduling leadership task {} to run every {} second(s)", task.getClass().getCanonicalName(), rate / 1000.0 ); this.futures.add(this.taskScheduler.scheduleAtFixedRate(task, rate)); break; case FIXED_DELAY: final long delay = task.getFixedDelay(); log.info( "Scheduling leadership task {} to run at a fixed delay of every {} second(s)", task.getClass().getCanonicalName(), delay / 1000.0 ); this.futures.add(this.taskScheduler.scheduleWithFixedDelay(task, delay)); break; default: log.error("Unknown Genie task type {}", task.getScheduleType()); } } ); } else if (leaderEvent instanceof OnRevokedEvent) { if (!this.isRunning) { return; } log.info("Leadership revoked."); this.isRunning = false; this.cancelTasks(); } else { log.warn("Unknown leadership event {}. Ignoring.", leaderEvent); } } }
LeaderTasksCoordinator { @EventListener public synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent) { if (leaderEvent instanceof OnGrantedEvent) { if (this.isRunning) { return; } log.info("Leadership granted."); this.isRunning = true; this.tasks.forEach( task -> { switch (task.getScheduleType()) { case TRIGGER: final Trigger trigger = task.getTrigger(); log.info( "Scheduling leadership task {} to run with trigger {}", task.getClass().getCanonicalName(), trigger ); this.futures.add(this.taskScheduler.schedule(task, trigger)); break; case FIXED_RATE: final long rate = task.getFixedRate(); log.info( "Scheduling leadership task {} to run every {} second(s)", task.getClass().getCanonicalName(), rate / 1000.0 ); this.futures.add(this.taskScheduler.scheduleAtFixedRate(task, rate)); break; case FIXED_DELAY: final long delay = task.getFixedDelay(); log.info( "Scheduling leadership task {} to run at a fixed delay of every {} second(s)", task.getClass().getCanonicalName(), delay / 1000.0 ); this.futures.add(this.taskScheduler.scheduleWithFixedDelay(task, delay)); break; default: log.error("Unknown Genie task type {}", task.getScheduleType()); } } ); } else if (leaderEvent instanceof OnRevokedEvent) { if (!this.isRunning) { return; } log.info("Leadership revoked."); this.isRunning = false; this.cancelTasks(); } else { log.warn("Unknown leadership event {}. Ignoring.", leaderEvent); } } LeaderTasksCoordinator(final TaskScheduler taskScheduler, final Collection<LeaderTask> tasks); }
LeaderTasksCoordinator { @EventListener public synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent) { if (leaderEvent instanceof OnGrantedEvent) { if (this.isRunning) { return; } log.info("Leadership granted."); this.isRunning = true; this.tasks.forEach( task -> { switch (task.getScheduleType()) { case TRIGGER: final Trigger trigger = task.getTrigger(); log.info( "Scheduling leadership task {} to run with trigger {}", task.getClass().getCanonicalName(), trigger ); this.futures.add(this.taskScheduler.schedule(task, trigger)); break; case FIXED_RATE: final long rate = task.getFixedRate(); log.info( "Scheduling leadership task {} to run every {} second(s)", task.getClass().getCanonicalName(), rate / 1000.0 ); this.futures.add(this.taskScheduler.scheduleAtFixedRate(task, rate)); break; case FIXED_DELAY: final long delay = task.getFixedDelay(); log.info( "Scheduling leadership task {} to run at a fixed delay of every {} second(s)", task.getClass().getCanonicalName(), delay / 1000.0 ); this.futures.add(this.taskScheduler.scheduleWithFixedDelay(task, delay)); break; default: log.error("Unknown Genie task type {}", task.getScheduleType()); } } ); } else if (leaderEvent instanceof OnRevokedEvent) { if (!this.isRunning) { return; } log.info("Leadership revoked."); this.isRunning = false; this.cancelTasks(); } else { log.warn("Unknown leadership event {}. Ignoring.", leaderEvent); } } LeaderTasksCoordinator(final TaskScheduler taskScheduler, final Collection<LeaderTask> tasks); @PreDestroy void preDestroy(); @EventListener synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent); }
LeaderTasksCoordinator { @EventListener public synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent) { if (leaderEvent instanceof OnGrantedEvent) { if (this.isRunning) { return; } log.info("Leadership granted."); this.isRunning = true; this.tasks.forEach( task -> { switch (task.getScheduleType()) { case TRIGGER: final Trigger trigger = task.getTrigger(); log.info( "Scheduling leadership task {} to run with trigger {}", task.getClass().getCanonicalName(), trigger ); this.futures.add(this.taskScheduler.schedule(task, trigger)); break; case FIXED_RATE: final long rate = task.getFixedRate(); log.info( "Scheduling leadership task {} to run every {} second(s)", task.getClass().getCanonicalName(), rate / 1000.0 ); this.futures.add(this.taskScheduler.scheduleAtFixedRate(task, rate)); break; case FIXED_DELAY: final long delay = task.getFixedDelay(); log.info( "Scheduling leadership task {} to run at a fixed delay of every {} second(s)", task.getClass().getCanonicalName(), delay / 1000.0 ); this.futures.add(this.taskScheduler.scheduleWithFixedDelay(task, delay)); break; default: log.error("Unknown Genie task type {}", task.getScheduleType()); } } ); } else if (leaderEvent instanceof OnRevokedEvent) { if (!this.isRunning) { return; } log.info("Leadership revoked."); this.isRunning = false; this.cancelTasks(); } else { log.warn("Unknown leadership event {}. Ignoring.", leaderEvent); } } LeaderTasksCoordinator(final TaskScheduler taskScheduler, final Collection<LeaderTask> tasks); @PreDestroy void preDestroy(); @EventListener synchronized void onLeaderEvent(final AbstractLeaderEvent leaderEvent);  }