QueueClient implements RequestQueue { @Override public Future<Void> deleteAllQueueItems(String queue, boolean unlock) { Future<Void> future = Future.future(); vertx.eventBus().send(getRedisquesAddress(), buildDeleteAllQueueItemsOperation(queue, unlock), (Handler<AsyncResult<Message<JsonObject>>>) event -> { if (event.succeeded()) { future.complete(); } else { future.fail("Failed to delete all queue items for queue " + queue + " with unlock " + unlock + ". Cause: " + event.cause()); } }); 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 testDeleteAllQueueItemsDoUnlock(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.deleteAllQueueItems, "myQueueToDeleteAndUnlock", true); message.reply(new JsonObject().put(STATUS, OK)); }); queueClient.deleteAllQueueItems("myQueueToDeleteAndUnlock", true).setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); }
@Test public void testDeleteAllQueueItemsDoNotUnlock(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.deleteAllQueueItems, "myQueueToDelete", false); message.reply(new JsonObject().put(STATUS, OK)); }); queueClient.deleteAllQueueItems("myQueueToDelete", false).setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); }
@Test public void testDeleteAllQueueItemsWithRedisquesReturningError(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.deleteAllQueueItems, "myQueueToDeleteAndUnlock", true); message.reply(new JsonObject().put(STATUS, ERROR)); }); queueClient.deleteAllQueueItems("myQueueToDeleteAndUnlock", true).setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); }
@Test public void testDeleteAllQueueItemsWithRedisquesFail(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.deleteAllQueueItems, "myQueueToDeleteAndUnlock", true); message.fail(0, "boom"); }); queueClient.deleteAllQueueItems("myQueueToDeleteAndUnlock", true).setHandler(event -> { context.assertTrue(event.failed()); context.assertTrue(event.cause().getMessage().contains("Failed to delete all queue items for queue myQueueToDeleteAndUnlock with unlock true")); async.complete(); }); }