@Test public void testFetchMessageQueues() { Set<MessageQueue> messageQueues = transactionBridge.fetchMessageQueues(TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC); assertThat(messageQueues.size()).isEqualTo(1); }
public Set<MessageQueue> fetchMessageQueues(String topic) { Set<MessageQueue> mqSet = new HashSet<>(); TopicConfig topicConfig = selectTopicConfig(topic); if (topicConfig != null && topicConfig.getReadQueueNums() > 0) { for (int i = 0; i < topicConfig.getReadQueueNums(); i++) { MessageQueue mq = new MessageQueue(); mq.setTopic(topic); mq.setBrokerName(brokerController.getBrokerConfig().getBrokerName()); mq.setQueueId(i); mqSet.add(mq); } } return mqSet; }
TransactionalMessageBridge { public Set<MessageQueue> fetchMessageQueues(String topic) { Set<MessageQueue> mqSet = new HashSet<>(); TopicConfig topicConfig = selectTopicConfig(topic); if (topicConfig != null && topicConfig.getReadQueueNums() > 0) { for (int i = 0; i < topicConfig.getReadQueueNums(); i++) { MessageQueue mq = new MessageQueue(); mq.setTopic(topic); mq.setBrokerName(brokerController.getBrokerConfig().getBrokerName()); mq.setQueueId(i); mqSet.add(mq); } } return mqSet; } }
TransactionalMessageBridge { public Set<MessageQueue> fetchMessageQueues(String topic) { Set<MessageQueue> mqSet = new HashSet<>(); TopicConfig topicConfig = selectTopicConfig(topic); if (topicConfig != null && topicConfig.getReadQueueNums() > 0) { for (int i = 0; i < topicConfig.getReadQueueNums(); i++) { MessageQueue mq = new MessageQueue(); mq.setTopic(topic); mq.setBrokerName(brokerController.getBrokerConfig().getBrokerName()); mq.setQueueId(i); mqSet.add(mq); } } return mqSet; } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); }
TransactionalMessageBridge { public Set<MessageQueue> fetchMessageQueues(String topic) { Set<MessageQueue> mqSet = new HashSet<>(); TopicConfig topicConfig = selectTopicConfig(topic); if (topicConfig != null && topicConfig.getReadQueueNums() > 0) { for (int i = 0; i < topicConfig.getReadQueueNums(); i++) { MessageQueue mq = new MessageQueue(); mq.setTopic(topic); mq.setBrokerName(brokerController.getBrokerConfig().getBrokerName()); mq.setQueueId(i); mqSet.add(mq); } } return mqSet; } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); long fetchConsumeOffset(MessageQueue mq); Set<MessageQueue> fetchMessageQueues(String topic); void updateConsumeOffset(MessageQueue mq, long offset); PullResult getHalfMessage(int queueId, long offset, int nums); PullResult getOpMessage(int queueId, long offset, int nums); PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner); CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner); boolean putOpMessage(MessageExt messageExt, String opType); PutMessageResult putMessageReturnResult(MessageExtBrokerInner messageInner); boolean putMessage(MessageExtBrokerInner messageInner); MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt); MessageExtBrokerInner renewHalfMessageInner(MessageExt msgExt); MessageExt lookMessageByOffset(final long commitLogOffset); BrokerController getBrokerController(); }
TransactionalMessageBridge { public Set<MessageQueue> fetchMessageQueues(String topic) { Set<MessageQueue> mqSet = new HashSet<>(); TopicConfig topicConfig = selectTopicConfig(topic); if (topicConfig != null && topicConfig.getReadQueueNums() > 0) { for (int i = 0; i < topicConfig.getReadQueueNums(); i++) { MessageQueue mq = new MessageQueue(); mq.setTopic(topic); mq.setBrokerName(brokerController.getBrokerConfig().getBrokerName()); mq.setQueueId(i); mqSet.add(mq); } } return mqSet; } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); long fetchConsumeOffset(MessageQueue mq); Set<MessageQueue> fetchMessageQueues(String topic); void updateConsumeOffset(MessageQueue mq, long offset); PullResult getHalfMessage(int queueId, long offset, int nums); PullResult getOpMessage(int queueId, long offset, int nums); PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner); CompletableFuture<PutMessageResult> asyncPutHalfMessage(MessageExtBrokerInner messageInner); boolean putOpMessage(MessageExt messageExt, String opType); PutMessageResult putMessageReturnResult(MessageExtBrokerInner messageInner); boolean putMessage(MessageExtBrokerInner messageInner); MessageExtBrokerInner renewImmunityHalfMessageInner(MessageExt msgExt); MessageExtBrokerInner renewHalfMessageInner(MessageExt msgExt); MessageExt lookMessageByOffset(final long commitLogOffset); BrokerController getBrokerController();  }