RedisReducedPropagationStorage implements ReducedPropagationStorage { @Override public Future<Boolean> addQueue(String queue, long expireTS) { Future<Boolean> future = Future.future(); List<String> keys = Collections.singletonList(QUEUE_TIMERS); List<String> arguments = Arrays.asList(queue, String.valueOf(expireTS)); StartQueueTimerRedisCommand cmd = new StartQueueTimerRedisCommand(startQueueTimerLuaScriptState, keys, arguments, redisClient, log, future); cmd.exec(0); 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 testAddQueue(TestContext context) { Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_TIMERS)); storage.addQueue("queue_1", 10).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(event.result()); context.assertTrue(jedis.exists(QUEUE_TIMERS)); Set<Tuple> expectedTuples = new HashSet<>(); expectedTuples.add(new Tuple("queue_1", 10.0)); assertQueuesTimersSetContent(1, expectedTuples); async.complete(); }); }
@Test public void testAddQueueNoDuplicates(TestContext context) { Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_TIMERS)); storage.addQueue("queue_1", 10).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(event.result()); context.assertTrue(jedis.exists(QUEUE_TIMERS)); Set<Tuple> expectedTuples = new HashSet<>(); expectedTuples.add(new Tuple("queue_1", 10.0)); assertQueuesTimersSetContent(1, expectedTuples); storage.addQueue("queue_1", 20).setHandler(event1 -> { context.assertTrue(event1.succeeded()); context.assertFalse(event1.result()); context.assertTrue(jedis.exists(QUEUE_TIMERS)); assertQueuesTimersSetContent(1, expectedTuples); async.complete(); }); }); }