package com.elab.mq.rocket.listener;

import com.elab.core.utils.StringUtils;
import com.elab.mq.rocket.exception.MsgProcessException;
import com.elab.mq.rocket.exception.MsgRecurringException;
import com.elab.mq.rocket.exception.MsgWaitException;
import com.elab.mq.rocket.msg.ConsumerDataStore;
import com.elab.mq.rocket.msg.entity.RmqConsumerEntity;
import com.elab.mq.rocket.utils.MsgConstants;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.util.Date;

/**
 * 公用抽象的消费者监听器
 *
 * @author ： liukx
 * @time ： 2019/7/18 - 13:54
 */
@RocketMQMessageListener(topic = "${rocketmq.producer.group}-topic", consumerGroup = "${rocketmq.producer.group}-group")
public class DefaultRocketMQListener implements RocketMQListener<MessageExt> {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired(required = false)
    private ConsumerDataStore consumerDataStore;
    @Value("${rocketmq.producer.group}")
    private String group;
    @Value("${spring.cloud.client.ipAddress:}")
    private String ip;
    @Autowired
    private TagInvokerListener tagInvokerListener;

    @Override
    public void onMessage(MessageExt messageExt) {
        RmqConsumerEntity entity = null;
        String topic = messageExt.getTopic();
        String keys = messageExt.getKeys();
        String msgId = messageExt.getMsgId();
        String tags = messageExt.getTags();
        try {
            if (StringUtils.isNotEmpty(keys)) {
                entity = consumerDataStore.getMsgKey(group, keys);
            } else {
                entity = consumerDataStore.getMsgId(group, msgId);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        boolean isInsert = false;
        // 1. 构建数据
        if (entity == null) {
            isInsert = true;
            entity = new RmqConsumerEntity();
            entity.setMsgId(msgId);
            entity.setConsuemrGroup(group);
            entity.setConsumerIp(ip);
            entity.setConsumerStatus(MsgConstants.MSG_OK);
            entity.setCreated(new Date());
            entity.setMsgKey(keys);
            entity.setRetryCount(0);
            entity.setTopic(topic);
        } else {
            // 如果存在消息记录，开始做状态比较
            Integer consumerStatus = entity.getConsumerStatus();
            if (MsgConstants.MSG_OK == consumerStatus) {
                logger.info(" topic : {} keys : {} id: {} 已经被消费过了.", topic, keys, msgId);
                // 已经消费成功了,不做处理
                return;
            } else if (MsgConstants.MSG_WAIT == consumerStatus) {
                // 还在处理中 ... 这个时候需要返回一个重新消费的状态
                throw new MsgWaitException(" 该 key [" + keys + "] 还在消费中 ... ");
            } else {
                int count = entity.getRetryCount() + 1;
                entity.setRetryCount(count);
                entity.setUpdated(new Date());
            }
        }

        String body = new String(messageExt.getBody());
        try {
            // 2. 具体执行消息
            tagInvokerListener.process(tags, body);
            // 3. 保存结果
            saveData(entity, isInsert);
        } catch (Exception e) {
            logger.error(" 业务消息消费失败 topic : " + topic + " key : " + keys, e);
            try {
                saveExceptionData(body, entity, isInsert);
            } catch (Exception ex) {
                logger.error(" 异常消费信息 存储数据库失败 ", ex);
                ex.printStackTrace();
            }
            throw new MsgProcessException(e.getMessage());
        }
    }

    private void saveData(RmqConsumerEntity entity, boolean isInsert) throws Exception {
        // 如果没有异常发生
        entity.setConsumerStatus(MsgConstants.MSG_OK);
        if (isInsert) {
            consumerDataStore.add(entity);
        } else {
            consumerDataStore.update(entity);
        }
        logger.info("保存消费消息数据. {}", entity);
    }

    private void saveExceptionData(String message, RmqConsumerEntity entity, boolean isInsert) throws Exception {
        if (isInsert) {
            entity.setConsumerStatus(MsgConstants.MSG_FAILS);
            consumerDataStore.add(entity);
            logger.info(" 添加失败消息保存成功 " + message);
            throw new MsgProcessException(" 消费消息执行失败 ... ");
        } else {
            consumerDataStore.update(entity);
            logger.info(" 修改失败消息保存成功 " + message);
            throw new MsgRecurringException(" 重试消息执行失败 ... ");
        }
    }
}
