package com.elab.mq.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import com.elab.core.utils.DateUtils;
import com.elab.core.utils.ObjectUtils;
import com.elab.mq.dao.IConsumerDao;
import com.elab.mq.model.ConsumerEntity;
import com.elab.mq.model.MessageModel;
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/4/22 - 18:02
 */
public abstract class AbstractMessageListener implements MessageListener {

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

    @Autowired
    private IConsumerDao consumerDao;

    @Value("${spring.application.name}")
    private String moduleName;

    private final String MQ_CONSUMER = "MQ_CONSUMER";


    /**
     * 客户端关注的topic消息,实现了该方法,消息监听被触发时会与之匹配。
     *
     * @return
     */
    public abstract String topic();

    /**
     * 客户端关注的topic下的tag子类标签
     *
     * @return
     */
    public String tag() {
        return "*";
    }


    private ConsumerEntity init(MessageModel message) throws Exception {
        String msgID = message.getMsgID();
        Integer producerId = 0;

        if (message.getProducerId() != null) {
            producerId = Integer.valueOf(message.getProducerId());
        }

        String topic = message.getTopic();
        String tag = message.getTag();
        ConsumerEntity oldConsumerEntity = new ConsumerEntity();
        ConsumerEntity consumerEntity = new ConsumerEntity();
        consumerEntity.setProducerId(producerId);
        consumerEntity.setTopicId(topic);
        consumerEntity.setModuleName(moduleName);
        consumerEntity.setStatus(1);
        consumerEntity.setMsgId(msgID);
        consumerEntity.setTag(tag);
        consumerEntity.setModuleMethod(getClass().getName());
        oldConsumerEntity = consumerDao.selectByObject(consumerEntity);
        if (oldConsumerEntity != null && oldConsumerEntity.getConsumerStatus().equals(1)) {
            logger.warn("消费者验证失败，已经消费成功过一次，不允许重复消费 消费编号 : " + oldConsumerEntity.getId());
            // throw new BusinessException("请不要重复消费，key：" + message);
            return oldConsumerEntity;
        }
        logger.debug("消费者验证通过，消费者消费的数据匹配上生产者数据，");
        if (oldConsumerEntity != null) {
            oldConsumerEntity.setRetryCount(oldConsumerEntity.getRetryCount() + 1);
            oldConsumerEntity.setConsumerStatus(-1);
            oldConsumerEntity.setUpdated(DateUtils.getCurrentDateTime());
            oldConsumerEntity.setCatId(Cat.getCurrentMessageId());
            consumerDao.updateById(oldConsumerEntity);
            logger.debug("触发消息重试 当前重试次数 : " + oldConsumerEntity.getRetryCount());
        } else {
            oldConsumerEntity = new ConsumerEntity();
            oldConsumerEntity.setProducerId(producerId);
            oldConsumerEntity.setContent(new String(message.getBody()));
            oldConsumerEntity.setHouseId(message.getHouseId());
            oldConsumerEntity.setModuleName(moduleName);
            oldConsumerEntity.setStatus(1);
            oldConsumerEntity.setCreated(DateUtils.getCurrentDateTime());
            oldConsumerEntity.setRetryCount(0);
            oldConsumerEntity.setTopicId(topic);
            oldConsumerEntity.setConsumerStatus(0);
            oldConsumerEntity.setMsgId(msgID);
            oldConsumerEntity.setTag(tag);
            oldConsumerEntity.setCatId(Cat.getCurrentMessageId());
            oldConsumerEntity.setModuleMethod(getClass().getName());
            int id = consumerDao.insert(oldConsumerEntity);
            oldConsumerEntity.setId(id);
            logger.debug(" 一条全新的消息，消息编号：" + id);
        }
        return oldConsumerEntity;
    }


    /**
     * 客户端的业务逻辑实现
     *
     * @param message        消息对象
     * @param consumeContext 消息上下文
     * @return
     */
    public abstract Action consume0(MessageModel message, ConsumeContext consumeContext) throws Exception;

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        String topic = message.getTopic();
        String tag = message.getTag();
        Transaction t = Cat.newTransaction(MQ_CONSUMER, topic + "_" + tag + "_" + getClass().getSimpleName());
        MessageModel messageModel = new MessageModel(message);
        logger.debug("消息处理被触发 : " + message.toString());
        Action action = null;
        try {
            ConsumerEntity oldConsumerEntity = init(messageModel);
            //如果已经成功过一次
            if (oldConsumerEntity.getConsumerStatus() == 1) {
                logger.warn(" 重复消费 过滤掉. 消费编号 : " + oldConsumerEntity.getId());
                t.setSuccessStatus();
                return Action.CommitMessage;
            }

            logger.debug("更新消费者数据： " + ObjectUtils.objectParseJsonStr(oldConsumerEntity));
            action = consume0(messageModel, consumeContext);

            if (action == null || Action.ReconsumeLater.equals(action)) {
                oldConsumerEntity.setConsumerStatus(-1);
                oldConsumerEntity.setUpdated(new Date());
                consumerDao.updateById(oldConsumerEntity);
            } else {
                oldConsumerEntity.setConsumerStatus(1);
                oldConsumerEntity.setUpdated(new Date());
                consumerDao.updateById(oldConsumerEntity);
            }
            t.setSuccessStatus();
        } catch (Exception e) {
            t.setStatus(e);
            logger.error("消息处理异常 : ", e);
            e.printStackTrace();
            return Action.ReconsumeLater;
        } finally {
            t.complete();
        }
        logger.debug("消息处理结果 : " + action);
        return action;
    }

}
