MySQLQueueDAO extends MySQLBaseDAO implements QueueDAO { @Override public void processUnacks(String queueName) { final String PROCESS_UNACKS = "UPDATE queue_message SET popped = false WHERE queue_name = ? AND popped = true AND TIMESTAMPADD(SECOND,60,CURRENT_TIMESTAMP) > deliver_on"; executeWithTransaction(PROCESS_UNACKS, q -> q.addParameter(queueName).executeUpdate()); } @Inject  MySQLQueueDAO(ObjectMapper om, DataSource ds); @Override void push(String queueName, String messageId, long offsetTimeInSecond); @Override void push(String queueName, String messageId, int priority, long offsetTimeInSecond); @Override void push(String queueName, List<Message> messages); @Override boolean pushIfNotExists(String queueName, String messageId, long offsetTimeInSecond); @Override boolean pushIfNotExists(String queueName, String messageId, int priority, long offsetTimeInSecond); @Override List<String> pop(String queueName, int count, int timeout); @Override List<Message> pollMessages(String queueName, int count, int timeout); @Override void remove(String queueName, String messageId); @Override int getSize(String queueName); @Override boolean ack(String queueName, String messageId); @Override boolean setUnackTimeout(String queueName, String messageId, long unackTimeout); @Override void flush(String queueName); @Override Map<String, Long> queuesDetail(); @Override Map<String, Map<String, Map<String, Long>>> queuesDetailVerbose(); void processAllUnacks(); @Override void processUnacks(String queueName); @Override boolean resetOffsetTime(String queueName, String messageId);  }
@Test public void processUnacksTest() { final String queueName = "process_unacks_test"; final int count = 10; final int unackedCount = 4; final String otherQueueName = "process_unacks_test_other_queue"; for(int i = 0; i < count; i++) { int offset = 0; if(i >= unackedCount){ offset = 1_000_000; } dao.push(queueName, "unack-" + i, offset); } for(int i = 0; i < count; i++) { dao.push(otherQueueName, "other-" + i, 0); } List<Message> polled = dao.pollMessages(queueName, 100, 10_000); assertNotNull(polled); assertFalse(polled.isEmpty()); assertEquals(unackedCount, polled.size()); dao.pollMessages(otherQueueName, 100, 10_000); assertTrue(dao.ack(queueName, "unack-1")); Long uacked = dao.queuesDetailVerbose().get(queueName).get("a").get("uacked"); assertNotNull(uacked); assertEquals(uacked.longValue(), unackedCount - 1); dao.processUnacks(queueName); Map<String, Map<String, Map<String, Long>>> details = dao.queuesDetailVerbose(); uacked = details.get(queueName).get("a").get("uacked"); assertNotNull(uacked); assertEquals("There should be no unacked messages", uacked.longValue(), 0); Long otherUacked = details.get(otherQueueName).get("a").get("uacked"); assertNotNull(otherUacked); assertEquals("Other queue should have unacked messages", otherUacked.longValue(), count); Long size = dao.queuesDetail().get(queueName); assertNotNull(size); assertEquals(size.longValue(), count - 1); }