public ConsumeConcurrentlyStatus consumeMessage()

in ons-core/ons-client/src/main/java/org/apache/rocketmq/ons/api/impl/rocketmq/BatchConsumerImpl.java [98:129]

30 lines of code
8 McCabe index (conditional complexity)


        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> rmqMsgList,
            ConsumeConcurrentlyContext contextRMQ) {
            List<Message> msgList = new ArrayList<Message>();
            for (MessageExt rmqMsg : rmqMsgList) {
                Message msg = ONSUtil.msgConvert(rmqMsg);
                Map<String, String> propertiesMap = rmqMsg.getProperties();
                msg.setMsgID(rmqMsg.getMsgId());
                if (propertiesMap != null && propertiesMap.get(Constants.TRANSACTION_ID) != null) {
                    msg.setMsgID(propertiesMap.get(Constants.TRANSACTION_ID));
                }
                msgList.add(msg);
            }

            BatchMessageListener listener = BatchConsumerImpl.this.subscribeTable.get(msgList.get(0).getTopic());
            if (null == listener) {
                throw new ONSClientException("BatchMessageListener is null");
            }

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