MySQLMessageQueueStorage implements MessageQueueStorage { @Override public FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize) { FetchResult result = new FetchResult(); try { final List<MessagePriority> dataObjs = m_msgDao.findIdAfter(tpp.getTopic(), tpp.getPartition(), tpp.getPriorityInt(), (Long) startOffset, batchSize, MessagePriorityEntity.READSET_FULL); long biggestOffset = 0L; if (dataObjs != null && !dataObjs.isEmpty()) { final TppConsumerMessageBatch batch = new TppConsumerMessageBatch(); for (MessagePriority dataObj : dataObjs) { biggestOffset = Math.max(biggestOffset, dataObj.getId()); batch.addMsgSeq(dataObj.getId(), 0); } final String topic = tpp.getTopic(); batch.setTopic(topic); batch.setPartition(tpp.getPartition()); batch.setPriority(tpp.isPriority()); batch.setTransferCallback(new TransferCallback() { @Override public void transfer(ByteBuf out) { for (MessagePriority dataObj : dataObjs) { PartialDecodedMessage partialMsg = new PartialDecodedMessage(); partialMsg.setRemainingRetries(0); partialMsg.setDurableProperties(Unpooled.wrappedBuffer(dataObj.getAttributes())); partialMsg.setBody(Unpooled.wrappedBuffer(dataObj.getPayload())); partialMsg.setBornTime(dataObj.getCreationDate().getTime()); partialMsg.setKey(dataObj.getRefKey()); partialMsg.setBodyCodecType(dataObj.getCodecType()); m_messageCodec.encode(partialMsg, out); } } }); batch.setResend(false); result.setBatch(batch); result.setOffset(biggestOffset); return result; } } catch (Exception e) { e.printStackTrace(); } return null; }  @Override void appendMessages(Tpp tpp, Collection<MessageRawDataBatch> batches); @Override synchronized Object findLastOffset(Tpp tpp, int groupId); @Override FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize); @Override void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, Integer>> msgSeqs); @Override void ack(Tpp tpp, String groupId, boolean resend, long msgSeq); @Override synchronized Object findLastResendOffset(Tpg tpg); @SuppressWarnings("unchecked") @Override FetchResult fetchResendMessages(Tpg tpg, Object startOffset, int batchSize);  }
@Test public void testFetchMessages() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); FetchResult result = s.fetchMessages(tpp, 0L, 10); ByteBuf out = Unpooled.buffer(); result.getBatch().getTransferCallback().transfer(out); assertTrue(out.readableBytes() > 0); assertTrue(!result.getBatch().getMsgSeqs().isEmpty()); }