RedisReducedPropagationStorage implements ReducedPropagationStorage { @Override public Future<Void> storeQueueRequest(String queue, JsonObject queueRequest) { Future<Void> future = Future.future(); if(StringUtils.isEmpty(queue)){ future.fail("Queue is not allowed to be empty"); return future; } if(queueRequest == null){ future.fail("Request is not allowed to be empty"); return future; } try { String queueRequestStr = queueRequest.encode(); redisClient.hset(QUEUE_REQUESTS, queue, queueRequestStr, reply -> { if(reply.failed()){ String message = "Failed to store request for queue '"+queue+"'. Cause: " + logCause(reply); log.error(message); future.fail(message); } else { future.complete(); } }); } catch (DecodeException ex){ future.fail("Failed to decode request for queue '"+queue+"'"); return future; } 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 testStoreQueueRequestInvalidQueueParam(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); storage.storeQueueRequest(null, new JsonObject()).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 testStoreQueueRequestInvalidRequestParam(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); storage.storeQueueRequest("queue_1", null).setHandler(event -> { context.assertFalse(event.succeeded()); context.assertEquals("Request is not allowed to be empty", event.cause().getMessage()); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); async.complete(); }); }
@Test public void testStoreQueueRequest(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_REQUESTS)); JsonObject request = new JsonObject().put("key1", 1234).put("key2", "abcd"); storage.storeQueueRequest("queue_1", request).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(jedis.exists(QUEUE_REQUESTS)); context.assertTrue(jedis.hexists(QUEUE_REQUESTS, "queue_1")); context.assertEquals(request.encode(), jedis.hget(QUEUE_REQUESTS, "queue_1")); async.complete(); }); }