public ConsumeConcurrentlyStatus consumeMessage()

in ons-core/ons-client/src/main/java/org/apache/rocketmq/ons/api/impl/rocketmq/ConsumerImpl.java [103:131]

27 lines of code
7 McCabe index (conditional complexity)


        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgsRMQList,
            ConsumeConcurrentlyContext contextRMQ) {
            MessageExt msgRMQ = msgsRMQList.get(0);
            Message msg = ONSUtil.msgConvert(msgRMQ);
            Map<String, String> stringStringMap = msgRMQ.getProperties();
            msg.setMsgID(msgRMQ.getMsgId());
            if (stringStringMap != null && stringStringMap.get(Constants.TRANSACTION_ID) != null) {
                msg.setMsgID(stringStringMap.get(Constants.TRANSACTION_ID));
            }
            MessageListener listener = ConsumerImpl.this.subscribeTable.get(msg.getTopic());
            if (null == listener) {
                throw new ONSClientException("MessageListener is null");
            }

            final ConsumeContext context = new ConsumeContext();
            Action action = listener.consume(msg, context);
            if (action != null) {
                switch (action) {
                    case CommitMessage:
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    case ReconsumeLater:
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    default:
                        break;
                }
            }

            return null;
        }