QueueClient implements RequestQueue { @Override public Future<Void> enqueueFuture(HttpRequest queuedRequest, String queue) { Future<Void> future = Future.future(); vertx.eventBus().send(getRedisquesAddress(), buildEnqueueOperation(queue, queuedRequest.toJsonObject().put(QUEUE_TIMESTAMP, System.currentTimeMillis()).encode()), (Handler<AsyncResult<Message<JsonObject>>>) event -> { if (OK.equals(event.result().body().getString(STATUS))) { monitoringHandler.updateLastUsedQueueSizeInformation(queue); monitoringHandler.updateEnqueue(); future.complete(); } else { future.fail(event.result().body().getString(MESSAGE)); } }); return future; } QueueClient(Vertx vertx, MonitoringHandler monitoringHandler); @Override void enqueue(final HttpServerRequest request, Buffer buffer, final String queue); @Override void enqueue(final HttpServerRequest request, MultiMap headers, Buffer buffer, final String queue); @Override void enqueue(HttpRequest request, final String queue); @Override void enqueue(HttpRequest request, final String queue, final Handler<Void> doneHandler); @Override void lockedEnqueue(HttpRequest queuedRequest, String queue, String lockRequestedBy, Handler<Void> doneHandler); @Override Future<Void> deleteLock(String queue); @Override Future<Void> deleteAllQueueItems(String queue, boolean unlock); @Override Future<Void> enqueueFuture(HttpRequest queuedRequest, String queue); static final String QUEUE_TIMESTAMP; static final Logger log; }
@Test public void testEnqueueFuture(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.enqueue, "myQueue"); message.reply(new JsonObject().put(STATUS, OK)); }); HttpRequest request = new HttpRequest(HttpMethod.PUT, "/targetUri", new CaseInsensitiveHeaders(), Buffer.buffer("{\"key\":\"value\"}").getBytes()); queueClient.enqueueFuture(request, "myQueue").setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); Mockito.verify(monitoringHandler, Mockito.timeout(1000).times(1)).updateLastUsedQueueSizeInformation(eq("myQueue")); Mockito.verify(monitoringHandler, Mockito.timeout(1000).times(1)).updateEnqueue(); }
@Test public void testEnqueueFutureNotUpdatingMonitoringHandlerOnRedisquesFail(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.enqueue, "myQueue"); message.reply(new JsonObject().put(STATUS, ERROR).put(MESSAGE, "enqueue:boom")); }); HttpRequest request = new HttpRequest(HttpMethod.PUT, "/targetUri", new CaseInsensitiveHeaders(), Buffer.buffer("{\"key\":\"value\"}").getBytes()); queueClient.enqueueFuture(request, "myQueue").setHandler(event -> { context.assertFalse(event.succeeded()); context.assertEquals("enqueue:boom", event.cause().getMessage()); async.complete(); }); Mockito.verifyZeroInteractions(monitoringHandler); }