@Test public void testConsumeMessageDirectly_WithNoException() { Map<ConsumeOrderlyStatus, CMResult> map = new HashMap(); map.put(ConsumeOrderlyStatus.SUCCESS, CMResult.CR_SUCCESS); map.put(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT, CMResult.CR_LATER); map.put(ConsumeOrderlyStatus.COMMIT, CMResult.CR_COMMIT); map.put(ConsumeOrderlyStatus.ROLLBACK, CMResult.CR_ROLLBACK); map.put(null, CMResult.CR_RETURN_NULL); for (ConsumeOrderlyStatus consumeOrderlyStatus : map.keySet()) { final ConsumeOrderlyStatus status = consumeOrderlyStatus; MessageListenerOrderly listenerOrderly = new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { return status; } }; ConsumeMessageOrderlyService consumeMessageOrderlyService = new ConsumeMessageOrderlyService(pushConsumer.getDefaultMQPushConsumerImpl(), listenerOrderly); MessageExt msg = new MessageExt(); msg.setTopic(topic); assertTrue(consumeMessageOrderlyService.consumeMessageDirectly(msg, brokerName).getConsumeResult().equals(map.get(consumeOrderlyStatus))); } }
@Override public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) { ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult(); result.setOrder(true); List<MessageExt> msgs = new ArrayList<MessageExt>(); msgs.add(msg); MessageQueue mq = new MessageQueue(); mq.setBrokerName(brokerName); mq.setTopic(msg.getTopic()); mq.setQueueId(msg.getQueueId()); ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq); this.defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, this.consumerGroup); final long beginTime = System.currentTimeMillis(); log.info("consumeMessageDirectly receive new message: {}", msg); try { ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context); if (status != null) { switch (status) { case COMMIT: result.setConsumeResult(CMResult.CR_COMMIT); break; case ROLLBACK: result.setConsumeResult(CMResult.CR_ROLLBACK); break; case SUCCESS: result.setConsumeResult(CMResult.CR_SUCCESS); break; case SUSPEND_CURRENT_QUEUE_A_MOMENT: result.setConsumeResult(CMResult.CR_LATER); break; default: break; } } else { result.setConsumeResult(CMResult.CR_RETURN_NULL); } } catch (Throwable e) { result.setConsumeResult(CMResult.CR_THROW_EXCEPTION); result.setRemark(RemotingHelper.exceptionSimpleDesc(e)); log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", RemotingHelper.exceptionSimpleDesc(e), ConsumeMessageOrderlyService.this.consumerGroup, msgs, mq), e); } result.setAutoCommit(context.isAutoCommit()); result.setSpentTimeMills(System.currentTimeMillis() - beginTime); log.info("consumeMessageDirectly Result: {}", result); return result; }
ConsumeMessageOrderlyService implements ConsumeMessageService { @Override public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) { ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult(); result.setOrder(true); List<MessageExt> msgs = new ArrayList<MessageExt>(); msgs.add(msg); MessageQueue mq = new MessageQueue(); mq.setBrokerName(brokerName); mq.setTopic(msg.getTopic()); mq.setQueueId(msg.getQueueId()); ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq); this.defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, this.consumerGroup); final long beginTime = System.currentTimeMillis(); log.info("consumeMessageDirectly receive new message: {}", msg); try { ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context); if (status != null) { switch (status) { case COMMIT: result.setConsumeResult(CMResult.CR_COMMIT); break; case ROLLBACK: result.setConsumeResult(CMResult.CR_ROLLBACK); break; case SUCCESS: result.setConsumeResult(CMResult.CR_SUCCESS); break; case SUSPEND_CURRENT_QUEUE_A_MOMENT: result.setConsumeResult(CMResult.CR_LATER); break; default: break; } } else { result.setConsumeResult(CMResult.CR_RETURN_NULL); } } catch (Throwable e) { result.setConsumeResult(CMResult.CR_THROW_EXCEPTION); result.setRemark(RemotingHelper.exceptionSimpleDesc(e)); log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", RemotingHelper.exceptionSimpleDesc(e), ConsumeMessageOrderlyService.this.consumerGroup, msgs, mq), e); } result.setAutoCommit(context.isAutoCommit()); result.setSpentTimeMills(System.currentTimeMillis() - beginTime); log.info("consumeMessageDirectly Result: {}", result); return result; } }
ConsumeMessageOrderlyService implements ConsumeMessageService { @Override public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) { ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult(); result.setOrder(true); List<MessageExt> msgs = new ArrayList<MessageExt>(); msgs.add(msg); MessageQueue mq = new MessageQueue(); mq.setBrokerName(brokerName); mq.setTopic(msg.getTopic()); mq.setQueueId(msg.getQueueId()); ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq); this.defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, this.consumerGroup); final long beginTime = System.currentTimeMillis(); log.info("consumeMessageDirectly receive new message: {}", msg); try { ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context); if (status != null) { switch (status) { case COMMIT: result.setConsumeResult(CMResult.CR_COMMIT); break; case ROLLBACK: result.setConsumeResult(CMResult.CR_ROLLBACK); break; case SUCCESS: result.setConsumeResult(CMResult.CR_SUCCESS); break; case SUSPEND_CURRENT_QUEUE_A_MOMENT: result.setConsumeResult(CMResult.CR_LATER); break; default: break; } } else { result.setConsumeResult(CMResult.CR_RETURN_NULL); } } catch (Throwable e) { result.setConsumeResult(CMResult.CR_THROW_EXCEPTION); result.setRemark(RemotingHelper.exceptionSimpleDesc(e)); log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", RemotingHelper.exceptionSimpleDesc(e), ConsumeMessageOrderlyService.this.consumerGroup, msgs, mq), e); } result.setAutoCommit(context.isAutoCommit()); result.setSpentTimeMills(System.currentTimeMillis() - beginTime); log.info("consumeMessageDirectly Result: {}", result); return result; } ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerOrderly messageListener); }
ConsumeMessageOrderlyService implements ConsumeMessageService { @Override public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) { ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult(); result.setOrder(true); List<MessageExt> msgs = new ArrayList<MessageExt>(); msgs.add(msg); MessageQueue mq = new MessageQueue(); mq.setBrokerName(brokerName); mq.setTopic(msg.getTopic()); mq.setQueueId(msg.getQueueId()); ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq); this.defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, this.consumerGroup); final long beginTime = System.currentTimeMillis(); log.info("consumeMessageDirectly receive new message: {}", msg); try { ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context); if (status != null) { switch (status) { case COMMIT: result.setConsumeResult(CMResult.CR_COMMIT); break; case ROLLBACK: result.setConsumeResult(CMResult.CR_ROLLBACK); break; case SUCCESS: result.setConsumeResult(CMResult.CR_SUCCESS); break; case SUSPEND_CURRENT_QUEUE_A_MOMENT: result.setConsumeResult(CMResult.CR_LATER); break; default: break; } } else { result.setConsumeResult(CMResult.CR_RETURN_NULL); } } catch (Throwable e) { result.setConsumeResult(CMResult.CR_THROW_EXCEPTION); result.setRemark(RemotingHelper.exceptionSimpleDesc(e)); log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", RemotingHelper.exceptionSimpleDesc(e), ConsumeMessageOrderlyService.this.consumerGroup, msgs, mq), e); } result.setAutoCommit(context.isAutoCommit()); result.setSpentTimeMills(System.currentTimeMillis() - beginTime); log.info("consumeMessageDirectly Result: {}", result); return result; } ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerOrderly messageListener); void start(); void shutdown(long awaitTerminateMillis); synchronized void unlockAllMQ(); @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 dispathToConsume); synchronized void lockMQPeriodically(); void tryLockLaterAndReconsume(final MessageQueue mq, final ProcessQueue processQueue,
        final long delayMills); synchronized boolean lockOneMQ(final MessageQueue mq); boolean processConsumeResult(
        final List<MessageExt> msgs,
        final ConsumeOrderlyStatus status,
        final ConsumeOrderlyContext context,
        final ConsumeRequest consumeRequest
    ); ConsumerStatsManager getConsumerStatsManager(); boolean sendMessageBack(final MessageExt msg); void resetNamespace(final List<MessageExt> msgs); }
ConsumeMessageOrderlyService implements ConsumeMessageService { @Override public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) { ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult(); result.setOrder(true); List<MessageExt> msgs = new ArrayList<MessageExt>(); msgs.add(msg); MessageQueue mq = new MessageQueue(); mq.setBrokerName(brokerName); mq.setTopic(msg.getTopic()); mq.setQueueId(msg.getQueueId()); ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq); this.defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, this.consumerGroup); final long beginTime = System.currentTimeMillis(); log.info("consumeMessageDirectly receive new message: {}", msg); try { ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context); if (status != null) { switch (status) { case COMMIT: result.setConsumeResult(CMResult.CR_COMMIT); break; case ROLLBACK: result.setConsumeResult(CMResult.CR_ROLLBACK); break; case SUCCESS: result.setConsumeResult(CMResult.CR_SUCCESS); break; case SUSPEND_CURRENT_QUEUE_A_MOMENT: result.setConsumeResult(CMResult.CR_LATER); break; default: break; } } else { result.setConsumeResult(CMResult.CR_RETURN_NULL); } } catch (Throwable e) { result.setConsumeResult(CMResult.CR_THROW_EXCEPTION); result.setRemark(RemotingHelper.exceptionSimpleDesc(e)); log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", RemotingHelper.exceptionSimpleDesc(e), ConsumeMessageOrderlyService.this.consumerGroup, msgs, mq), e); } result.setAutoCommit(context.isAutoCommit()); result.setSpentTimeMills(System.currentTimeMillis() - beginTime); log.info("consumeMessageDirectly Result: {}", result); return result; } ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerOrderly messageListener); void start(); void shutdown(long awaitTerminateMillis); synchronized void unlockAllMQ(); @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 dispathToConsume); synchronized void lockMQPeriodically(); void tryLockLaterAndReconsume(final MessageQueue mq, final ProcessQueue processQueue,
        final long delayMills); synchronized boolean lockOneMQ(final MessageQueue mq); boolean processConsumeResult(
        final List<MessageExt> msgs,
        final ConsumeOrderlyStatus status,
        final ConsumeOrderlyContext context,
        final ConsumeRequest consumeRequest
    ); ConsumerStatsManager getConsumerStatsManager(); boolean sendMessageBack(final MessageExt msg); void resetNamespace(final List<MessageExt> msgs);  }