### Question:
RqueueTaskAggregatorService implements ApplicationListener<RqueueExecutionEvent>, DisposableBean, SmartLifecycle { @Override public void onApplicationEvent(RqueueExecutionEvent event) { synchronized (aggregatorLock) { if (log.isTraceEnabled()) { log.trace("Event {}", event); } QueueDetail queueDetail = (QueueDetail) event.getSource(); String queueName = queueDetail.getName(); QueueEvents queueEvents = queueNameToEvents.get(queueName); if (queueEvents == null) { queueEvents = new QueueEvents(event); } else { queueEvents.addEvent(event); } if (processingRequired(queueEvents)) { if (log.isTraceEnabled()) { log.trace("Adding events to the queue"); } queue.add(queueEvents); queueNameToEvents.remove(queueName); } else { queueNameToEvents.put(queueName, queueEvents); } aggregatorLock.notifyAll(); } } @Autowired  RqueueTaskAggregatorService(
      RqueueConfig rqueueConfig,
      RqueueWebConfig rqueueWebConfig,
      RqueueLockManager rqueueLockManager,
      RqueueQStatsDao rqueueQStatsDao); @Override void destroy(); @Override void start(); @Override void stop(); @Override boolean isRunning(); @Override void onApplicationEvent(RqueueExecutionEvent event);  }### Answer:
@Test public void onApplicationEvent() throws TimedOutException { if (LocalDateTime.now(ZoneOffset.UTC).getHour() == 23) { log.info("This test cannot be run at this time"); return; } String id = "__rq::q-stat::" + queueName; doReturn(id).when(rqueueConfig).getQueueStatisticsKey(queueName); doReturn("__rq::lock::" + id).when(rqueueConfig).getLockKey(id); doReturn(true) .when(rqueueLockManager) .acquireLock( "__rq::lock::" + id, Duration.ofSeconds(Constants.AGGREGATION_LOCK_DURATION_IN_SECONDS)); List<QueueStatistics> queueStatistics = new ArrayList<>(); doAnswer( invocation -> { queueStatistics.add(invocation.getArgument(0)); return null; }) .when(rqueueQStatsDao) .save(any()); RqueueExecutionEvent event; TasksStat tasksStat = new TasksStat(); int totalEvents = 0; for (; totalEvents < 498; totalEvents++) { event = generateTaskEvent(); addEvent(event, tasksStat, true); } if (tasksStat.movedToDlq == 0) { totalEvents += 1; event = generateTaskEventWithStatus(TaskStatus.MOVED_TO_DLQ); addEvent(event, tasksStat, true); } if (tasksStat.success == 0) { totalEvents += 1; event = generateTaskEventWithStatus(TaskStatus.SUCCESSFUL); addEvent(event, tasksStat, true); } if (tasksStat.discarded == 0) { totalEvents += 1; event = generateTaskEventWithStatus(TaskStatus.DISCARDED); addEvent(event, tasksStat, totalEvents < 500); } if (tasksStat.retried == 0) { totalEvents += 1; event = generateTaskEventWithStatus(TaskStatus.DISCARDED); event.getRqueueMessage().setFailureCount(10); addEvent(event, tasksStat, totalEvents < 500); } for (; totalEvents < 501; totalEvents++) { event = generateTaskEvent(); addEvent(event, tasksStat, totalEvents < 500); } TimeoutUtils.waitFor( () -> !queueStatistics.isEmpty(), 60 * Constants.ONE_MILLI, "stats to be saved."); QueueStatistics statistics = queueStatistics.get(0); String date = DateTimeUtils.today().toString(); assertEquals(statistics.getId(), id); QueueStatisticsTest.validate(statistics, 1); QueueStatisticsTest.checkNonNull(statistics, date); assertEquals(tasksStat.jobRunTime(), statistics.jobRunTime(date)); assertEquals(tasksStat.discarded, statistics.tasksDiscarded(date)); assertEquals(tasksStat.success, statistics.tasksSuccessful(date)); assertEquals(tasksStat.movedToDlq, statistics.tasksMovedToDeadLetter(date)); assertEquals(tasksStat.retried, statistics.tasksRetried(date)); }