@Ignore @Test public void testPullMessage_ConsumeSuccess() throws InterruptedException, RemotingException, MQBrokerException, NoSuchFieldException,Exception { final CountDownLatch countDownLatch = new CountDownLatch(1); final MessageExt[] messageExts = new MessageExt[1]; ConsumeMessageConcurrentlyService normalServie = new ConsumeMessageConcurrentlyService(pushConsumer.getDefaultMQPushConsumerImpl(), new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { messageExts[0] = msgs.get(0); countDownLatch.countDown(); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(normalServie); PullMessageService pullMessageService = mQClientFactory.getPullMessageService(); pullMessageService.executePullRequestImmediately(createPullRequest()); countDownLatch.await(); Thread.sleep(1000); org.apache.rocketmq.common.protocol.body.ConsumeStatus stats = normalServie.getConsumerStatsManager().consumeStatus(pushConsumer.getDefaultMQPushConsumerImpl().groupName(),topic); ConsumerStatsManager mgr = normalServie.getConsumerStatsManager(); Field statItmeSetField = mgr.getClass().getDeclaredField("topicAndGroupConsumeOKTPS"); statItmeSetField.setAccessible(true); StatsItemSet itemSet = (StatsItemSet)statItmeSetField.get(mgr); StatsItem item = itemSet.getAndCreateStatsItem(topic + "@" + pushConsumer.getDefaultMQPushConsumerImpl().groupName()); assertThat(item.getValue().get()).isGreaterThan(0L); assertThat(messageExts[0].getTopic()).isEqualTo(topic); assertThat(messageExts[0].getBody()).isEqualTo(new byte[] {'a'}); }
public ConsumerStatsManager getConsumerStatsManager() { return this.defaultMQPushConsumerImpl.getConsumerStatsManager(); }
ConsumeMessageConcurrentlyService implements ConsumeMessageService { public ConsumerStatsManager getConsumerStatsManager() { return this.defaultMQPushConsumerImpl.getConsumerStatsManager(); } }
ConsumeMessageConcurrentlyService implements ConsumeMessageService { public ConsumerStatsManager getConsumerStatsManager() { return this.defaultMQPushConsumerImpl.getConsumerStatsManager(); } ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerConcurrently messageListener); }
ConsumeMessageConcurrentlyService implements ConsumeMessageService { public ConsumerStatsManager getConsumerStatsManager() { return this.defaultMQPushConsumerImpl.getConsumerStatsManager(); } ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerConcurrently messageListener); void start(); void shutdown(long awaitTerminateMillis); @Override void updateCorePoolSize(int corePoolSize); @Override void incCorePoolSize(); @Override void decCorePoolSize(); @Override int getCorePoolSize(); @Override ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName); @Override void submitConsumeRequest(
        final List<MessageExt> msgs,
        final ProcessQueue processQueue,
        final MessageQueue messageQueue,
        final boolean dispatchToConsume); void processConsumeResult(
        final ConsumeConcurrentlyStatus status,
        final ConsumeConcurrentlyContext context,
        final ConsumeRequest consumeRequest
    ); ConsumerStatsManager getConsumerStatsManager(); boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context); }
ConsumeMessageConcurrentlyService implements ConsumeMessageService { public ConsumerStatsManager getConsumerStatsManager() { return this.defaultMQPushConsumerImpl.getConsumerStatsManager(); } ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerConcurrently messageListener); void start(); void shutdown(long awaitTerminateMillis); @Override void updateCorePoolSize(int corePoolSize); @Override void incCorePoolSize(); @Override void decCorePoolSize(); @Override int getCorePoolSize(); @Override ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName); @Override void submitConsumeRequest(
        final List<MessageExt> msgs,
        final ProcessQueue processQueue,
        final MessageQueue messageQueue,
        final boolean dispatchToConsume); void processConsumeResult(
        final ConsumeConcurrentlyStatus status,
        final ConsumeConcurrentlyContext context,
        final ConsumeRequest consumeRequest
    ); ConsumerStatsManager getConsumerStatsManager(); boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context);  }