QueueClient implements RequestQueue { @Override public Future<Void> deleteLock(String queue) { Future<Void> future = Future.future(); vertx.eventBus().send(getRedisquesAddress(), buildDeleteLockOperation(queue), (Handler<AsyncResult<Message<JsonObject>>>) event -> { if (event.failed()) { future.fail(event.cause()); return; } if (OK.equals(event.result().body().getString(STATUS))) { future.complete(); return; } future.fail("Failed to delete lock for queue " + queue); }); 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 testDeletelock(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.deleteLock, "myQueueToUnlock"); message.reply(new JsonObject().put(STATUS, OK)); }); queueClient.deleteLock("myQueueToUnlock").setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); }
@Test public void testDeleteLockWithRedisquesFail(TestContext context){ Async async = context.async(); vertx.eventBus().localConsumer(Address.redisquesAddress(), (Handler<Message<JsonObject>>) message -> { validateMessage(context, message, QueueOperation.deleteLock, "myQueueToUnlock"); message.reply(new JsonObject().put(STATUS, ERROR)); }); queueClient.deleteLock("myQueueToUnlock").setHandler(event -> { context.assertTrue(event.failed()); context.assertTrue(event.cause().getMessage().contains("Failed to delete lock for queue myQueueToUnlock")); async.complete(); }); }