@Test public void testGetHalfMessage() { when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), ArgumentMatchers.nullable(MessageFilter.class))).thenReturn(createGetMessageResult(GetMessageStatus.NO_MESSAGE_IN_QUEUE)); PullResult result = transactionBridge.getHalfMessage(0, 0, 1); assertThat(result.getPullStatus()).isEqualTo(PullStatus.NO_NEW_MSG); }
public PullResult getHalfMessage(int queueId, long offset, int nums) { String group = TransactionalMessageUtil.buildConsumerGroup(); String topic = TransactionalMessageUtil.buildHalfTopic(); SubscriptionData sub = new SubscriptionData(topic, "*"); return getMessage(group, topic, queueId, offset, nums, sub); }
TransactionalMessageBridge { public PullResult getHalfMessage(int queueId, long offset, int nums) { String group = TransactionalMessageUtil.buildConsumerGroup(); String topic = TransactionalMessageUtil.buildHalfTopic(); SubscriptionData sub = new SubscriptionData(topic, "*"); return getMessage(group, topic, queueId, offset, nums, sub); } }
TransactionalMessageBridge { public PullResult getHalfMessage(int queueId, long offset, int nums) { String group = TransactionalMessageUtil.buildConsumerGroup(); String topic = TransactionalMessageUtil.buildHalfTopic(); SubscriptionData sub = new SubscriptionData(topic, "*"); return getMessage(group, topic, queueId, offset, nums, sub); } TransactionalMessageBridge(BrokerController brokerController, MessageStore store); }
TransactionalMessageBridge { public PullResult getHalfMessage(int queueId, long offset, int nums) { String group = TransactionalMessageUtil.buildConsumerGroup(); String topic = TransactionalMessageUtil.buildHalfTopic(); SubscriptionData sub = new SubscriptionData(topic, "*"); return getMessage(group, topic, queueId, offset, nums, sub); } 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 PullResult getHalfMessage(int queueId, long offset, int nums) { String group = TransactionalMessageUtil.buildConsumerGroup(); String topic = TransactionalMessageUtil.buildHalfTopic(); SubscriptionData sub = new SubscriptionData(topic, "*"); return getMessage(group, topic, queueId, offset, nums, sub); } 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();  }