QueueClient implements RequestQueue { @Override public void lockedEnqueue(HttpRequest queuedRequest, String queue, String lockRequestedBy, Handler<Void> doneHandler) { vertx.eventBus().send(getRedisquesAddress(), buildLockedEnqueueOperation(queue, queuedRequest.toJsonObject().put(QUEUE_TIMESTAMP, System.currentTimeMillis()).encode(), lockRequestedBy), (Handler<AsyncResult<Message<JsonObject>>>) event -> { if (OK.equals(event.result().body().getString(STATUS))) { monitoringHandler.updateLastUsedQueueSizeInformation(queue); monitoringHandler.updateEnqueue(); } if (doneHandler != null) { doneHandler.handle(null); } }); } 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 testLockedEnqueue(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.lockedEnqueue, "myQueue", "LockRequester"); message.reply(new JsonObject().put(STATUS, OK)); }); HttpRequest request = new HttpRequest(HttpMethod.PUT, "/targetUri", new CaseInsensitiveHeaders(), Buffer.buffer("{\"key\":\"value\"}").getBytes()); queueClient.lockedEnqueue(request, "myQueue", "LockRequester", event -> async.complete()); Mockito.verify(monitoringHandler, Mockito.timeout(1000).times(1)).updateEnqueue(); Mockito.verify(monitoringHandler, Mockito.timeout(1000).times(1)).updateLastUsedQueueSizeInformation(eq("myQueue")); }
@Test public void testLockedEnqueueNotUpdatingMonitoringHandlerOnRedisquesFail(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.lockedEnqueue, "myQueue", "LockRequester"); message.reply(new JsonObject().put(STATUS, ERROR)); }); HttpRequest request = new HttpRequest(HttpMethod.PUT, "/targetUri", new CaseInsensitiveHeaders(), Buffer.buffer("{\"key\":\"value\"}").getBytes()); queueClient.lockedEnqueue(request, "myQueue", "LockRequester", event -> async.complete()); Mockito.verifyZeroInteractions(monitoringHandler); }