RedisReducedPropagationStorage implements ReducedPropagationStorage { @Override public Future<JsonObject> getQueueRequest(String queue) { Future<JsonObject> future = Future.future(); if(StringUtils.isEmpty(queue)){ future.fail("Queue is not allowed to be empty"); return future; } redisClient.hget(QUEUE_REQUESTS, queue, reply -> { if(reply.failed()){ String message = "get queue request '"+queue+"' from hash '" + QUEUE_REQUESTS+ "' resulted in cause " + logCause(reply); log.error(message); future.fail(message); } else { String resultStr = reply.result(); if(StringUtils.isNotEmpty(resultStr)) { try { JsonObject queueRequest = new JsonObject(resultStr); future.complete(queueRequest); } catch (DecodeException ex) { future.fail("Failed to decode queue request for queue '" + queue + "'. Got this from storage: " + resultStr); } } else { future.complete(null); } } }); 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 testGetQueueRequestInvalidParam(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); storage.getQueueRequest(null).setHandler(event -> { context.assertFalse(event.succeeded()); context.assertEquals("Queue is not allowed to be empty", event.cause().getMessage()); async.complete(); }); }
@Test public void testGetQueueRequestNotExisting(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); storage.getQueueRequest("queue_1").setHandler(event -> { context.assertTrue(event.succeeded()); context.assertNull(event.result()); async.complete(); }); }
@Test @Ignore public void testGetQueueRequest(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); JsonObject expected = new JsonObject().put("myKey", 12345); jedis.hset(QUEUE_REQUESTS, "queue_1", expected.encode()); storage.getQueueRequest("queue_1").setHandler(event -> { context.assertTrue(event.succeeded()); context.assertEquals(expected, event.result()); async.complete(); }); }
@Test public void testGetQueueRequestInvalidJson(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); jedis.hset(QUEUE_REQUESTS, "queue_1", "not_a_json_value"); storage.getQueueRequest("queue_1").setHandler(event -> { context.assertFalse(event.succeeded()); context.assertEquals("Failed to decode queue request for queue 'queue_1'. Got this from storage: not_a_json_value", event.cause().getMessage()); async.complete(); }); }