ReducedPropagationManager { public void startExpiredQueueProcessing(long intervalMs) { log.info("About to start periodic processing of expired queues with an interval of " + intervalMs + " ms"); vertx.cancelTimer(processExpiredQueuesTimerId); processExpiredQueuesTimerId = vertx.setPeriodic(intervalMs, event -> { final String token = createToken("reducedpropagation_expired_queue_processing"); acquireLock(this.lock, PROCESS_EXPIRED_QUEUES_LOCK, token, getLockExpiry(intervalMs), log).setHandler(lockEvent -> { if(lockEvent.succeeded()){ if(lockEvent.result()){ processExpiredQueues(token); } } else { log.error("Could not acquire lock '"+PROCESS_EXPIRED_QUEUES_LOCK+"'. Message: " + lockEvent.cause().getMessage()); } }); }); } ReducedPropagationManager(Vertx vertx, ReducedPropagationStorage storage, RequestQueue requestQueue, Lock lock); void startExpiredQueueProcessing(long intervalMs); Future<Void> processIncomingRequest(HttpMethod method, String targetUri, MultiMap queueHeaders, Buffer payload, String queue, long propagationIntervalMs, Handler<Void> doneHandler); static final String PROCESS_EXPIRED_QUEUES_LOCK; static final String LOCK_REQUESTER; static final String PROCESSOR_ADDRESS; static final String MANAGER_QUEUE_PREFIX; }
@Test public void testStartExpiredQueueProcessing(TestContext context) { Mockito.when(lock.acquireLock(anyString(), anyString(), anyLong())).thenReturn(Future.succeededFuture(Boolean.TRUE)); Mockito.when(lock.releaseLock(anyString(), anyString())).thenReturn(Future.succeededFuture(Boolean.TRUE)); Mockito.when(reducedPropagationStorage.removeExpiredQueues(anyLong())) .thenReturn(Future.succeededFuture(new ArrayList<>())); manager.startExpiredQueueProcessing(10); verify(reducedPropagationStorage, timeout(110).atLeast(10)).removeExpiredQueues(anyLong()); }
@Test public void testExpiredQueueProcessingNotExecutedWhenLocked(TestContext context) { Mockito.when(lock.acquireLock(anyString(), anyString(), anyLong())).thenReturn(Future.succeededFuture(Boolean.FALSE)); Mockito.when(lock.releaseLock(anyString(), anyString())).thenReturn(Future.succeededFuture(Boolean.TRUE)); Mockito.when(reducedPropagationStorage.removeExpiredQueues(anyLong())) .thenReturn(Future.succeededFuture(new ArrayList<>())); manager.startExpiredQueueProcessing(10); verify(reducedPropagationStorage, timeout(110).never()).removeExpiredQueues(anyLong()); }