QueueCircuitBreakerImpl implements QueueCircuitBreaker, RuleChangesObserver, Refreshable { @Override public Future<Void> lockQueue(String queueName, HttpRequest queuedRequest) { Future<Void> future = Future.future(); PatternAndCircuitHash patternAndCircuitHash = getPatternAndCircuitHashFromRequest(queuedRequest); if (patternAndCircuitHash != null) { queueCircuitBreakerStorage.lockQueue(queueName, patternAndCircuitHash).setHandler(event -> { if (event.failed()) { future.fail(event.cause()); return; } vertx.eventBus().send(redisquesAddress, buildPutLockOperation(queueName, "queue_circuit_breaker"), (Handler<AsyncResult<Message<JsonObject>>>) reply -> { if (reply.failed()) { future.fail(reply.cause()); return; } if (OK.equals(reply.result().body().getString(STATUS))) { log.info("locked queue '" + queueName + "' because the circuit '" + patternAndCircuitHash.getPattern().pattern() + "' is open"); future.complete(); } else { future.fail("failed to lock queue '" + queueName + "'. Queue should have been locked, because the circuit '" + patternAndCircuitHash.getPattern().pattern() + "' is open"); } }); }); } else { failWithNoRuleToCircuitMappingMessage(future, queueName, queuedRequest); } 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 testQueueLock(TestContext context){ Async async = context.async(2); HttpRequest req = new HttpRequest(HttpMethod.PUT, "/playground/circuitBreaker/test", MultiMap.caseInsensitiveMultiMap(), null); Mockito.when(ruleToCircuitMapping.getCircuitFromRequestUri(anyString())) .thenReturn(new PatternAndCircuitHash(Pattern.compile("/someCircuit"), "someCircuitHash")); Mockito.when(queueCircuitBreakerStorage.lockQueue(anyString(), any(PatternAndCircuitHash.class))) .thenReturn(Future.succeededFuture()); vertx.eventBus().consumer(Address.redisquesAddress(), (Message<JsonObject> event) -> { context.assertEquals("putLock", event.body().getString("operation")); event.reply(new JsonObject().put("status", "ok")); async.countDown(); }); queueCircuitBreaker.lockQueue("someQueue", req).setHandler(event -> { context.assertTrue(event.succeeded()); verify(queueCircuitBreakerStorage, times(1)).lockQueue(anyString(), any(PatternAndCircuitHash.class)); async.countDown(); }); async.awaitSuccess(); }
@Test public void testQueueLockFailingRedisques(TestContext context){ Async async = context.async(2); HttpRequest req = new HttpRequest(HttpMethod.PUT, "/playground/circuitBreaker/test", MultiMap.caseInsensitiveMultiMap(), null); Mockito.when(ruleToCircuitMapping.getCircuitFromRequestUri(anyString())) .thenReturn(new PatternAndCircuitHash(Pattern.compile("/someCircuit"), "someCircuitHash")); Mockito.when(queueCircuitBreakerStorage.lockQueue(anyString(), any(PatternAndCircuitHash.class))) .thenReturn(Future.succeededFuture()); vertx.eventBus().consumer(Address.redisquesAddress(), (Message<JsonObject> event) -> { context.assertEquals("putLock", event.body().getString("operation")); event.reply(new JsonObject().put("status", "error")); async.countDown(); }); queueCircuitBreaker.lockQueue("someQueue", req).setHandler(event -> { context.assertTrue(event.failed()); context.assertTrue(event.cause().getMessage().contains("failed to lock queue 'someQueue'")); verify(queueCircuitBreakerStorage, times(1)).lockQueue(anyString(), any(PatternAndCircuitHash.class)); async.countDown(); }); async.awaitSuccess(); }
@Test public void testQueueLockFailingStorage(TestContext context){ Async async = context.async(); HttpRequest req = new HttpRequest(HttpMethod.PUT, "/playground/circuitBreaker/test", MultiMap.caseInsensitiveMultiMap(), null); Mockito.when(ruleToCircuitMapping.getCircuitFromRequestUri(anyString())) .thenReturn(new PatternAndCircuitHash(Pattern.compile("/someCircuit"), "someCircuitHash")); Mockito.when(queueCircuitBreakerStorage.lockQueue(anyString(), any(PatternAndCircuitHash.class))) .thenReturn(Future.failedFuture("queue could not be locked")); vertx.eventBus().consumer(Address.redisquesAddress(), (Message<JsonObject> event) -> { context.fail("Redisques should not have been called when the storage failed"); }); queueCircuitBreaker.lockQueue("someQueue", req).setHandler(event -> { context.assertTrue(event.failed()); context.assertTrue(event.cause().getMessage().contains("queue could not be locked")); verify(queueCircuitBreakerStorage, times(1)).lockQueue(anyString(), any(PatternAndCircuitHash.class)); async.complete(); }); async.awaitSuccess(); }