RedisReducedPropagationStorage implements ReducedPropagationStorage { @Override public Future<Void> removeQueueRequest(String queue) { Future<Void> future = Future.future(); if(StringUtils.isEmpty(queue)){ future.fail("Queue is not allowed to be empty"); return future; } redisClient.hdel(QUEUE_REQUESTS, queue, reply -> { if(reply.failed()){ String message = "Failed to remove request for queue '"+queue+"'. Cause: " + logCause(reply); log.error(message); future.fail(message); } else { future.complete(); } }); return future; } RedisReducedPropagationStorage(RedisClient redisClient); @Override Future<List<String>> removeExpiredQueues(long currentTS); @Override Future<Boolean> addQueue(String queue, long expireTS); @Override Future<Void> storeQueueRequest(String queue, JsonObject queueRequest); @Override Future<JsonObject> getQueueRequest(String queue); @Override Future<Void> removeQueueRequest(String queue);  }
@Test public void testRemoveQueueRequestInvalidQueueParam(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); storage.removeQueueRequest(null).setHandler(event -> { context.assertFalse(event.succeeded()); context.assertEquals("Queue is not allowed to be empty", event.cause().getMessage()); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); async.complete(); }); }
@Test public void testRemoveQueueRequestHashNotExisting(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); storage.removeQueueRequest("not_existing_queue").setHandler(event -> { context.assertTrue(event.succeeded()); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); async.complete(); }); }
@Test public void testRemoveQueueRequestNotExisting(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); jedis.hset(QUEUE_REQUESTS, "some_other_queue", new JsonObject().encode()); storage.removeQueueRequest("not_existing_queue").setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(jedis.exists(QUEUE_REQUESTS)); context.assertTrue(jedis.hexists(QUEUE_REQUESTS, "some_other_queue")); async.complete(); }); }
@Test public void testRemoveQueueRequest(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); jedis.hset(QUEUE_REQUESTS, "some_other_queue", new JsonObject().encode()); jedis.hset(QUEUE_REQUESTS, "queue_1", new JsonObject().encode()); context.assertTrue(jedis.hexists(QUEUE_REQUESTS, "queue_1")); storage.removeQueueRequest("queue_1").setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(jedis.exists(QUEUE_REQUESTS)); context.assertTrue(jedis.hexists(QUEUE_REQUESTS, "some_other_queue")); context.assertFalse(jedis.hexists(QUEUE_REQUESTS, "queue_1")); async.complete(); }); }