QueueCircuitBreakerImpl implements QueueCircuitBreaker, RuleChangesObserver, Refreshable { @Override public Future<Long> unlockSampleQueues() { log.info("About to unlock a sample queue for each circuit"); Future<Long> future = Future.future(); queueCircuitBreakerStorage.unlockSampleQueues().setHandler(event -> { if (event.failed()) { future.fail(event.cause().getMessage()); return; } List<String> queuesToUnlock = event.result(); if (queuesToUnlock == null || queuesToUnlock.isEmpty()) { future.complete(0L); return; } final AtomicInteger futureCounter = new AtomicInteger(queuesToUnlock.size()); List<String> failedFutures = new ArrayList<>(); for (String queueToUnlock : queuesToUnlock) { log.info("About to unlock sample queue '" + queueToUnlock + "'"); unlockQueue(queueToUnlock).setHandler(event1 -> { futureCounter.decrementAndGet(); if (event1.failed()) { failedFutures.add(event1.cause().getMessage()); } if (futureCounter.get() == 0) { if (failedFutures.size() > 0) { future.fail("The following queues could not be unlocked: " + failedFutures); } else { future.complete((long) queuesToUnlock.size()); } } }); } }); return future; } QueueCircuitBreakerImpl(Vertx vertx, Lock lock, String redisquesAddress, QueueCircuitBreakerStorage queueCircuitBreakerStorage, RuleProvider ruleProvider, QueueCircuitBreakerRulePatternToCircuitMapping ruleToCircuitMapping, QueueCircuitBreakerConfigurationResourceManager configResourceManager, Handler<HttpServerRequest> queueCircuitBreakerHttpRequestHandler, int requestHandlerPort); @Override void rulesChanged(List<Rule> rules); @Override void refresh(); @Override boolean isCircuitCheckEnabled(); @Override boolean isStatisticsUpdateEnabled(); @Override Future<QueueCircuitState> handleQueuedRequest(String queueName, HttpRequest queuedRequest); @Override Future<Void> updateStatistics(String queueName, HttpRequest queuedRequest, QueueResponseType queueResponseType); @Override Future<Void> closeCircuit(HttpRequest queuedRequest); @Override Future<Void> closeAllCircuits(); @Override Future<Void> reOpenCircuit(HttpRequest queuedRequest); @Override Future<Void> lockQueue(String queueName, HttpRequest queuedRequest); @Override Future<String> unlockNextQueue(); @Override Future<Long> setOpenCircuitsToHalfOpen(); @Override Future<Long> unlockSampleQueues(); @Override Future<String> unlockQueue(String queueName); static final String OPEN_TO_HALF_OPEN_TASK_LOCK; static final String UNLOCK_QUEUES_TASK_LOCK; static final String UNLOCK_SAMPLE_QUEUES_TASK_LOCK; }
@Test public void testUnlockSampleQueues(TestContext context){ Async async = context.async(4); Mockito.when(queueCircuitBreakerStorage.unlockSampleQueues()) .thenReturn(Future.succeededFuture(Arrays.asList("q1", "q2", "q3"))); vertx.eventBus().consumer(Address.redisquesAddress(), (Message<JsonObject> event) -> { context.assertEquals("deleteLock", event.body().getString("operation")); event.reply(new JsonObject().put("status", "ok")); async.countDown(); }); queueCircuitBreaker.unlockSampleQueues().setHandler(event -> { context.assertTrue(event.succeeded()); context.assertEquals(3L, event.result()); async.countDown(); }); async.awaitSuccess(); }
@Test public void testUnlockSampleQueuesFailingUnlock(TestContext context){ Async async = context.async(4); Mockito.when(queueCircuitBreakerStorage.unlockSampleQueues()) .thenReturn(Future.succeededFuture(Arrays.asList("q1", "q2", "q3"))); vertx.eventBus().consumer(Address.redisquesAddress(), (Message<JsonObject> event) -> { context.assertEquals("deleteLock", event.body().getString("operation")); if(event.body().getJsonObject("payload").getString("queuename").equalsIgnoreCase("q2")){ event.reply(new JsonObject().put("status", "error")); } else { event.reply(new JsonObject().put("status", "ok")); } async.countDown(); }); queueCircuitBreaker.unlockSampleQueues().setHandler(event -> { context.assertTrue(event.failed()); context.assertTrue(event.cause().getMessage().contains("The following queues could not be unlocked: [q2]")); async.countDown(); }); async.awaitSuccess(); }
@Test public void testUnlockSampleQueuesFailingStorage(TestContext context){ Async async = context.async(); Mockito.when(queueCircuitBreakerStorage.unlockSampleQueues()) .thenReturn(Future.failedFuture("unable to lock queues")); vertx.eventBus().consumer(Address.redisquesAddress(), (Message<JsonObject> event) -> { context.fail("Redisques should not have been called when the storage failed"); }); queueCircuitBreaker.unlockSampleQueues().setHandler(event -> { context.assertTrue(event.failed()); context.assertTrue(event.cause().getMessage().contains("unable to lock queues")); async.complete(); }); async.awaitSuccess(); }