@Test public void testPullMessageAsync_Success() throws Exception { doAnswer(new Answer() { @Override public Object answer(InvocationOnMock mock) throws Throwable { PullMessageRequestHeader requestHeader = mock.getArgument(1); PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.singletonList(new MessageExt())); PullCallback pullCallback = mock.getArgument(4); pullCallback.onSuccess(pullResult); return null; } }).when(mQClientAPIImpl).pullMessage(anyString(), any(PullMessageRequestHeader.class), anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)); MessageQueue messageQueue = new MessageQueue(topic, brokerName, 0); pullConsumer.pull(messageQueue, "*", 1024, 3, new PullCallback() { @Override public void onSuccess(PullResult pullResult) { assertThat(pullResult).isNotNull(); assertThat(pullResult.getPullStatus()).isEqualTo(PullStatus.FOUND); assertThat(pullResult.getNextBeginOffset()).isEqualTo(1024 + 1); assertThat(pullResult.getMinOffset()).isEqualTo(123); assertThat(pullResult.getMaxOffset()).isEqualTo(2048); assertThat(pullResult.getMsgFoundList()).isEqualTo(new ArrayList<Object>()); } @Override public void onException(Throwable e) { } }); }
@Override public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums); }
DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer { @Override public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums); } }
DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer { @Override public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums); } DefaultMQPullConsumer();  DefaultMQPullConsumer(final String consumerGroup, RPCHook rpcHook);  DefaultMQPullConsumer(final String consumerGroup);  DefaultMQPullConsumer(RPCHook rpcHook); }
DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer { @Override public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums); } DefaultMQPullConsumer();  DefaultMQPullConsumer(final String consumerGroup, RPCHook rpcHook);  DefaultMQPullConsumer(final String consumerGroup);  DefaultMQPullConsumer(RPCHook rpcHook); @Override void createTopic(String key, String newTopic, int queueNum); @Override void createTopic(String key, String newTopic, int queueNum, int topicSysFlag); @Override long searchOffset(MessageQueue mq, long timestamp); @Override long maxOffset(MessageQueue mq); @Override long minOffset(MessageQueue mq); @Override long earliestMsgStoreTime(MessageQueue mq); @Override MessageExt viewMessage(String offsetMsgId); @Override QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end); AllocateMessageQueueStrategy getAllocateMessageQueueStrategy(); void setAllocateMessageQueueStrategy(AllocateMessageQueueStrategy allocateMessageQueueStrategy); long getBrokerSuspendMaxTimeMillis(); void setBrokerSuspendMaxTimeMillis(long brokerSuspendMaxTimeMillis); String getConsumerGroup(); void setConsumerGroup(String consumerGroup); long getConsumerPullTimeoutMillis(); void setConsumerPullTimeoutMillis(long consumerPullTimeoutMillis); long getConsumerTimeoutMillisWhenSuspend(); void setConsumerTimeoutMillisWhenSuspend(long consumerTimeoutMillisWhenSuspend); MessageModel getMessageModel(); void setMessageModel(MessageModel messageModel); MessageQueueListener getMessageQueueListener(); void setMessageQueueListener(MessageQueueListener messageQueueListener); Set<String> getRegisterTopics(); void setRegisterTopics(Set<String> registerTopics); @Override void sendMessageBack(MessageExt msg, int delayLevel); @Override void sendMessageBack(MessageExt msg, int delayLevel, String brokerName); @Override Set<MessageQueue> fetchSubscribeMessageQueues(String topic); @Override void start(); @Override void shutdown(); @Override void registerMessageQueueListener(String topic, MessageQueueListener listener); @Override PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums); @Override PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums, long timeout); @Override void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback); @Override void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback,
        long timeout); @Override PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums); @Override void pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums,
        PullCallback pullCallback); @Override void updateConsumeOffset(MessageQueue mq, long offset); @Override long fetchConsumeOffset(MessageQueue mq, boolean fromStore); @Override Set<MessageQueue> fetchMessageQueuesInBalance(String topic); @Override MessageExt viewMessage(String topic,
        String uniqKey); @Override void sendMessageBack(MessageExt msg, int delayLevel, String brokerName, String consumerGroup); OffsetStore getOffsetStore(); void setOffsetStore(OffsetStore offsetStore); DefaultMQPullConsumerImpl getDefaultMQPullConsumerImpl(); boolean isUnitMode(); void setUnitMode(boolean isUnitMode); int getMaxReconsumeTimes(); void setMaxReconsumeTimes(final int maxReconsumeTimes); }
DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer { @Override public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { return this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums); } DefaultMQPullConsumer();  DefaultMQPullConsumer(final String consumerGroup, RPCHook rpcHook);  DefaultMQPullConsumer(final String consumerGroup);  DefaultMQPullConsumer(RPCHook rpcHook); @Override void createTopic(String key, String newTopic, int queueNum); @Override void createTopic(String key, String newTopic, int queueNum, int topicSysFlag); @Override long searchOffset(MessageQueue mq, long timestamp); @Override long maxOffset(MessageQueue mq); @Override long minOffset(MessageQueue mq); @Override long earliestMsgStoreTime(MessageQueue mq); @Override MessageExt viewMessage(String offsetMsgId); @Override QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end); AllocateMessageQueueStrategy getAllocateMessageQueueStrategy(); void setAllocateMessageQueueStrategy(AllocateMessageQueueStrategy allocateMessageQueueStrategy); long getBrokerSuspendMaxTimeMillis(); void setBrokerSuspendMaxTimeMillis(long brokerSuspendMaxTimeMillis); String getConsumerGroup(); void setConsumerGroup(String consumerGroup); long getConsumerPullTimeoutMillis(); void setConsumerPullTimeoutMillis(long consumerPullTimeoutMillis); long getConsumerTimeoutMillisWhenSuspend(); void setConsumerTimeoutMillisWhenSuspend(long consumerTimeoutMillisWhenSuspend); MessageModel getMessageModel(); void setMessageModel(MessageModel messageModel); MessageQueueListener getMessageQueueListener(); void setMessageQueueListener(MessageQueueListener messageQueueListener); Set<String> getRegisterTopics(); void setRegisterTopics(Set<String> registerTopics); @Override void sendMessageBack(MessageExt msg, int delayLevel); @Override void sendMessageBack(MessageExt msg, int delayLevel, String brokerName); @Override Set<MessageQueue> fetchSubscribeMessageQueues(String topic); @Override void start(); @Override void shutdown(); @Override void registerMessageQueueListener(String topic, MessageQueueListener listener); @Override PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums); @Override PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums, long timeout); @Override void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback); @Override void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback,
        long timeout); @Override PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums); @Override void pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums,
        PullCallback pullCallback); @Override void updateConsumeOffset(MessageQueue mq, long offset); @Override long fetchConsumeOffset(MessageQueue mq, boolean fromStore); @Override Set<MessageQueue> fetchMessageQueuesInBalance(String topic); @Override MessageExt viewMessage(String topic,
        String uniqKey); @Override void sendMessageBack(MessageExt msg, int delayLevel, String brokerName, String consumerGroup); OffsetStore getOffsetStore(); void setOffsetStore(OffsetStore offsetStore); DefaultMQPullConsumerImpl getDefaultMQPullConsumerImpl(); boolean isUnitMode(); void setUnitMode(boolean isUnitMode); int getMaxReconsumeTimes(); void setMaxReconsumeTimes(final int maxReconsumeTimes);  }