QueueProcessor { private boolean isCircuitCheckEnabled() { return queueCircuitBreaker != null && queueCircuitBreaker.isCircuitCheckEnabled(); } QueueProcessor(final Vertx vertx, final HttpClient httpClient, final MonitoringHandler monitoringHandler);  QueueProcessor(final Vertx vertx, final HttpClient httpClient, final MonitoringHandler monitoringHandler, QueueCircuitBreaker queueCircuitBreaker);  QueueProcessor(final Vertx vertx, final HttpClient httpClient, final MonitoringHandler monitoringHandler,
                          QueueCircuitBreaker queueCircuitBreaker, boolean immediatelyStartQueueProcessing); void startQueueProcessing(); void stopQueueProcessing(); boolean isQueueProcessingStarted(); String getQueueProcessorAddress(); static boolean httpMethodIsQueueable(HttpMethod method); HttpClient getHttpClient();  }
@Test public void testOpenCircuit(TestContext context){ Async async = context.async(); QueueCircuitBreaker circuitBreaker = Mockito.spy(new ConfigurableQueueCircuitBreaker(QueueCircuitState.OPEN, true, true)); new QueueProcessor(vertx, httpClient, monitoringHandler, circuitBreaker); vertx.eventBus().send(Address.queueProcessorAddress(), buildQueueEventBusMessage("my_queue"), event -> { context.assertTrue(event.succeeded()); JsonObject result = (JsonObject) event.result().body(); context.assertEquals("error", result.getString("status")); context.assertTrue(result.getString("message").contains("Circuit for queue my_queue is OPEN")); verify(httpClient, never()).request(any(HttpMethod.class), anyString(), Matchers.<Handler<HttpClientResponse>>any()); verify(circuitBreaker, times(1)).isCircuitCheckEnabled(); verify(circuitBreaker, times(1)).handleQueuedRequest(anyString(), any(HttpRequest.class)); async.complete(); }); }
@Test public void testInactiveQueueCircuitBreaker(TestContext context){ Async async = context.async(); QueueCircuitBreaker circuitBreaker = Mockito.spy(new ConfigurableQueueCircuitBreaker(QueueCircuitState.OPEN, false, true)); new QueueProcessor(vertx, httpClient, monitoringHandler, circuitBreaker); vertx.eventBus().send(Address.queueProcessorAddress(), buildQueueEventBusMessage("my_queue"), new DeliveryOptions().setSendTimeout(1000), event -> { verify(circuitBreaker, times(1)).isCircuitCheckEnabled(); verify(circuitBreaker, never()).handleQueuedRequest(anyString(), any(HttpRequest.class)); verify(httpClient, times(1)).request(any(HttpMethod.class), anyString(), Matchers.<Handler<HttpClientResponse>>any()); async.complete(); }); }