@Test public void testFetchMessages() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); FetchResult result = s.fetchMessages(tpp, 0L, 10, null); ByteBuf out = Unpooled.buffer(); result.getBatch().getTransferCallback().transfer(out); assertTrue(out.readableBytes() > 0); assertTrue(!result.getBatch().getMessageMetas().isEmpty()); }
@Override public FetchResult fetchMessages(Tpp tpp, List<Object> offsets) { List<MessagePriority> msgs = new ArrayList<MessagePriority>(); if (!hasStorageForPriority(tpp.getTopic(), tpp.getPriorityInt())) { return buildFetchResult(tpp, msgs, null); } for (Long[] subOffsets : splitOffsets(offsets)) { try { msgs.addAll(m_msgDao.findWithOffsets(tpp.getTopic(), tpp.getPartition(), tpp.getPriorityInt(), subOffsets, MessagePriorityEntity.READSET_FULL)); } catch (Exception e) { log.error("Failed to fetch message({}).", tpp, e); continue; } } return buildFetchResult(tpp, msgs, null); }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public FetchResult fetchMessages(Tpp tpp, List<Object> offsets) { List<MessagePriority> msgs = new ArrayList<MessagePriority>(); if (!hasStorageForPriority(tpp.getTopic(), tpp.getPriorityInt())) { return buildFetchResult(tpp, msgs, null); } for (Long[] subOffsets : splitOffsets(offsets)) { try { msgs.addAll(m_msgDao.findWithOffsets(tpp.getTopic(), tpp.getPartition(), tpp.getPriorityInt(), subOffsets, MessagePriorityEntity.READSET_FULL)); } catch (Exception e) { log.error("Failed to fetch message({}).", tpp, e); continue; } } return buildFetchResult(tpp, msgs, null); } }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public FetchResult fetchMessages(Tpp tpp, List<Object> offsets) { List<MessagePriority> msgs = new ArrayList<MessagePriority>(); if (!hasStorageForPriority(tpp.getTopic(), tpp.getPriorityInt())) { return buildFetchResult(tpp, msgs, null); } for (Long[] subOffsets : splitOffsets(offsets)) { try { msgs.addAll(m_msgDao.findWithOffsets(tpp.getTopic(), tpp.getPartition(), tpp.getPriorityInt(), subOffsets, MessagePriorityEntity.READSET_FULL)); } catch (Exception e) { log.error("Failed to fetch message({}).", tpp, e); continue; } } return buildFetchResult(tpp, msgs, null); }  }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public FetchResult fetchMessages(Tpp tpp, List<Object> offsets) { List<MessagePriority> msgs = new ArrayList<MessagePriority>(); if (!hasStorageForPriority(tpp.getTopic(), tpp.getPriorityInt())) { return buildFetchResult(tpp, msgs, null); } for (Long[] subOffsets : splitOffsets(offsets)) { try { msgs.addAll(m_msgDao.findWithOffsets(tpp.getTopic(), tpp.getPartition(), tpp.getPriorityInt(), subOffsets, MessagePriorityEntity.READSET_FULL)); } catch (Exception e) { log.error("Failed to fetch message({}).", tpp, e); continue; } } return buildFetchResult(tpp, msgs, null); }  @Override void appendMessages(String topicName, int partition, boolean priority,
	      Collection<MessageBatchWithRawData> batches); @Override synchronized Object findLastOffset(Tpp tpp, int groupId); @Override FetchResult fetchMessages(Tpp tpp, List<Object> offsets); @Override FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize, String filter); @Override void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, MessageMeta>> msgId2Metas); @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); @Override Object findMessageOffsetByTime(Tpp tpp, long time); @Override void initialize(); }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public FetchResult fetchMessages(Tpp tpp, List<Object> offsets) { List<MessagePriority> msgs = new ArrayList<MessagePriority>(); if (!hasStorageForPriority(tpp.getTopic(), tpp.getPriorityInt())) { return buildFetchResult(tpp, msgs, null); } for (Long[] subOffsets : splitOffsets(offsets)) { try { msgs.addAll(m_msgDao.findWithOffsets(tpp.getTopic(), tpp.getPartition(), tpp.getPriorityInt(), subOffsets, MessagePriorityEntity.READSET_FULL)); } catch (Exception e) { log.error("Failed to fetch message({}).", tpp, e); continue; } } return buildFetchResult(tpp, msgs, null); }  @Override void appendMessages(String topicName, int partition, boolean priority,
	      Collection<MessageBatchWithRawData> batches); @Override synchronized Object findLastOffset(Tpp tpp, int groupId); @Override FetchResult fetchMessages(Tpp tpp, List<Object> offsets); @Override FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize, String filter); @Override void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, MessageMeta>> msgId2Metas); @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); @Override Object findMessageOffsetByTime(Tpp tpp, long time); @Override void initialize();  }