RedisReducedPropagationStorage implements ReducedPropagationStorage { @Override public Future<List<String>> removeExpiredQueues(long currentTS) { Future<List<String>> future = Future.future(); List<String> keys = Collections.singletonList(QUEUE_TIMERS); List<String> arguments = Collections.singletonList(String.valueOf(currentTS)); RemoveExpiredQueuesRedisCommand cmd = new RemoveExpiredQueuesRedisCommand(removeExpiredQueuesRedisCommand, 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 testRemoveExpiredQueuesNoExpiredQueuesFound(TestContext context) { Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_TIMERS)); jedis.zadd(QUEUE_TIMERS, 15, "queue_1"); jedis.zadd(QUEUE_TIMERS, 20, "queue_2"); jedis.zadd(QUEUE_TIMERS, 25, "queue_3"); storage.removeExpiredQueues(10).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertNotNull(event.result()); context.assertEquals(Collections.emptyList(), event.result()); Set<Tuple> expected = new HashSet<>(); expected.add(new Tuple("queue_1", 15.0)); expected.add(new Tuple("queue_2", 20.0)); expected.add(new Tuple("queue_3", 25.0)); assertQueuesTimersSetContent(3, expected); async.complete(); }); }
@Test public void testRemoveExpiredQueuesExpiredQueuesFound(TestContext context) { Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_TIMERS)); jedis.zadd(QUEUE_TIMERS, 15, "queue_1"); jedis.zadd(QUEUE_TIMERS, 20, "queue_2"); jedis.zadd(QUEUE_TIMERS, 25, "queue_3"); jedis.zadd(QUEUE_TIMERS, 50, "queue_4"); jedis.zadd(QUEUE_TIMERS, 75, "queue_5"); storage.removeExpiredQueues(20).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertNotNull(event.result()); context.assertEquals(Arrays.asList("queue_1", "queue_2"), event.result()); Set<Tuple> expected = new HashSet<>(); expected.add(new Tuple("queue_3", 25.0)); expected.add(new Tuple("queue_4", 50.0)); expected.add(new Tuple("queue_5", 75.0)); assertQueuesTimersSetContent(3, expected); async.complete(); }); }
@Test public void testRemoveExpiredQueuesAllQueuesExpired(TestContext context) { Async async = context.async(); context.assertFalse(jedis.exists(QUEUE_TIMERS)); jedis.zadd(QUEUE_TIMERS, 15, "queue_1"); jedis.zadd(QUEUE_TIMERS, 20, "queue_2"); jedis.zadd(QUEUE_TIMERS, 25, "queue_3"); storage.removeExpiredQueues(50).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertNotNull(event.result()); context.assertEquals(Arrays.asList("queue_1", "queue_2", "queue_3"), event.result()); assertQueuesTimersSetContent(0, Collections.emptySet()); async.complete(); }); }