EventSchedulerService { public void start() { synchronized (this) { if (scheduledFuture == null || scheduledFuture.isDone()) { scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> { try { triggerEvents(); } catch(Exception e) { logger.error("Error while running triggerEvents", e); } }, initialDelay, batchReadInterval, TimeUnit.MILLISECONDS); logger.info("EventSchedulerService started"); } } } @Inject  EventSchedulerService(EventSchedulerDao eventSchedulerDao, EventSchedulerRegistry eventSchedulerRegistry,
                                 @Named("eventScheduler.batchRead.intervalms") Integer batchReadInterval,
                                 @Named("eventScheduler.batchRead.batchSize") Integer batchSize,
                                 ObjectMapper objectMapper); void start(); void stop(); Boolean isRunning(); void setInitialDelay(Long initialDelay);  }
@Test public void testTriggerEvent_shouldTriggerEventWhenOldEventFound() throws Exception { long now = System.currentTimeMillis()/1000; when(eventSchedulerDao.retrieveOldest(batchSize)). thenReturn(Arrays.asList(new ScheduledEvent("smCorId", "event_name1", now-2, "{\"name\":\"event_name1\",\"type\":\"some_type\",\"data\":\"data\",\"eventSource\":\"internal\"}"), new ScheduledEvent("smCorId", "event_name2", now + 10000, "{\"name\":\"event_name2\",\"type\":\"some_type\",\"data\":\"data\",\"eventSource\":\"internal\"}"))). thenReturn(Arrays.asList(new ScheduledEvent("smCorId", "event_name2", now + 10000, "{\"name\":\"event_name2\",\"type\":\"some_type\",\"data\":\"data\",\"eventSource\":\"internal\"}"))); eventSchedulerService.start(); Thread.sleep(300); verify(eventSchedulerRegistry).triggerEvent("event_name1", "data", "smCorId", "internal"); verify(eventSchedulerRegistry).deregisterEvent("smCorId", "event_name1"); Thread.sleep(1100); verifyNoMoreInteractions(eventSchedulerRegistry); }
@Test public void testNoTriggerEvents_shouldNotTriggerEventsWithScheduledTimeOfFuture() throws Exception { long now = System.currentTimeMillis()/1000; when(eventSchedulerDao.retrieveOldest(batchSize)). thenReturn(Arrays.asList(new ScheduledEvent("smCorId", "event_name1", now + 9, "{\"name\":\"event_name1\",\"type\":\"some_type\",\"data\":\"data\",\"eventSource\":\"internal\"}"), new ScheduledEvent("smCorId", "event_name2", now + 10, "{\"name\":\"event_name2\",\"type\":\"some_type\",\"data\":\"data\",\"eventSource\":\"internal\"}"))); eventSchedulerService.start(); Thread.sleep(1000); verifyZeroInteractions(eventSchedulerRegistry); }