RedisDynoQueue implements DynoQueue { @Override public boolean setTimeout(String messageId, long timeout) { return execute("setTimeout", "(a shard in) " + queueName, () -> { String json = nonQuorumConn.hget(messageStoreKey, messageId); if (json == null) { return false; } Message message = om.readValue(json, Message.class); message.setTimeout(timeout); for (String shard : allShards) { String queueShard = getQueueShardKey(queueName, shard); Double score = quorumConn.zscore(queueShard, messageId); if (score != null) { double priorityd = message.getPriority() / 100; double newScore = Long.valueOf(clock.millis() + timeout).doubleValue() + priorityd; ZAddParams params = ZAddParams.zAddParams().xx(); quorumConn.zadd(queueShard, newScore, messageId, params); json = om.writeValueAsString(message); quorumConn.hset(messageStoreKey, message.getId(), json); return true; } } return false; }); } RedisDynoQueue(String redisKeyPrefix, String queueName, Set<String> allShards, String shardName, ShardingStrategy shardingStrategy, boolean singleRingTopology);  RedisDynoQueue(String redisKeyPrefix, String queueName, Set<String> allShards, String shardName, int unackScheduleInMS, ShardingStrategy shardingStrategy, boolean singleRingTopology);  RedisDynoQueue(Clock clock, String redisKeyPrefix, String queueName, Set<String> allShards, String shardName, int unackScheduleInMS, ShardingStrategy shardingStrategy, boolean singleRingTopology); RedisDynoQueue withQuorumConn(JedisCommands quorumConn); RedisDynoQueue withNonQuorumConn(JedisCommands nonQuorumConn); RedisDynoQueue withUnackTime(int unackTime); @Override String getName(); @Override int getUnackTime(); @Override List<String> push(final List<Message> messages); @Override List<Message> peek(final int messageCount); @Override List<Message> unsafePeekAllShards(final int messageCount); @Override List<Message> pop(int messageCount, int wait, TimeUnit unit); @Override Message popWithMsgId(String messageId); @Override Message unsafePopWithMsgIdAllShards(String messageId); Message popWithMsgIdHelper(String messageId, String targetShard, boolean warnIfNotExists); List<Message> unsafePopAllShards(int messageCount, int wait, TimeUnit unit); @Override boolean ack(String messageId); @Override void ack(List<Message> messages); @Override boolean setUnackTimeout(String messageId, long timeout); @Override boolean setTimeout(String messageId, long timeout); @Override boolean remove(String messageId); @Override boolean atomicRemove(String messageId); @Override boolean ensure(Message message); @Override boolean containsPredicate(String predicate); @Override String getMsgWithPredicate(String predicate); @Override boolean containsPredicate(String predicate, boolean localShardOnly); @Override String getMsgWithPredicate(String predicate, boolean localShardOnly); @Override Message popMsgWithPredicate(String predicate, boolean localShardOnly); @Override List<Message> bulkPop(int messageCount, int wait, TimeUnit unit); @Override List<Message> unsafeBulkPop(int messageCount, int wait, TimeUnit unit); @Override Message get(String messageId); @Override List<Message> getAllMessages(); @Override Message localGet(String messageId); @Override long size(); @Override Map<String, Map<String, Long>> shardSizes(); @Override void clear(); @Override void processUnacks(); @Override List<Message> findStaleMessages(); @Override void atomicProcessUnacks(); @Override void close();  }
@Test public void testSetTimeout() { rdq.clear(); Message msg = new Message("x001", "Hello World"); msg.setPriority(3); msg.setTimeout(20_000); rdq.push(Arrays.asList(msg)); List<Message> popped = rdq.pop(1, 1, TimeUnit.SECONDS); assertTrue(popped.isEmpty()); boolean updated = rdq.setTimeout(msg.getId(), 1); assertTrue(updated); popped = rdq.pop(1, 1, TimeUnit.SECONDS); assertEquals(1, popped.size()); assertEquals(1, popped.get(0).getTimeout()); updated = rdq.setTimeout(msg.getId(), 1); assertTrue(!updated); }