PostgresQueueDAO extends PostgresBaseDAO implements QueueDAO { @Override public List<Message> pollMessages(String queueName, int count, int timeout) { List<Message> messages = getWithTransactionWithOutErrorPropagation(tx -> popMessages(tx, queueName, count, timeout)); if(messages == null) return new ArrayList<>(); return messages; } @Inject  PostgresQueueDAO(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 pollMessagesTest() { final List<Message> messages = new ArrayList<>(); final String queueName = "issue399_testQueue"; final int totalSize = 10; for(int i = 0; i < totalSize; i++) { String payload = "{\"id\": " + i + ", \"msg\":\"test " + i + "\"}"; Message m = new Message("testmsg-" + i, payload, ""); if (i % 2 == 0) { m.setPriority(99-i); } messages.add(m); } dao.push(queueName, ImmutableList.copyOf(messages)); assertEquals("Queue size mismatch", totalSize, dao.getSize(queueName)); final int firstPollSize = 3; List<Message> firstPoll = dao.pollMessages(queueName, firstPollSize, 10_000); assertNotNull("First poll was null", firstPoll); assertFalse("First poll was empty", firstPoll.isEmpty()); assertEquals("First poll size mismatch", firstPollSize, firstPoll.size()); final int secondPollSize = 4; List<Message> secondPoll = dao.pollMessages(queueName, secondPollSize, 10_000); assertNotNull("Second poll was null", secondPoll); assertFalse("Second poll was empty", secondPoll.isEmpty()); assertEquals("Second poll size mismatch", secondPollSize, secondPoll.size()); assertEquals("Total queue size should have remained the same", totalSize, dao.getSize(queueName)); final long expectedSize = totalSize - firstPollSize - secondPollSize; try(Connection c = testUtil.getDataSource().getConnection()) { String UNPOPPED = "SELECT COUNT(*) FROM queue_message WHERE queue_name = ? AND popped = false"; try(Query q = new Query(testUtil.getObjectMapper(), c, UNPOPPED)) { long count = q.addParameter(queueName).executeCount(); assertEquals("Remaining queue size mismatch", expectedSize, count); } } catch (Exception ex) { fail(ex.getMessage()); } }