package com.elab.spring.intercept;

import com.alibaba.fastjson.JSON;
import com.dianping.cat.Cat;
import com.elab.core.utils.DataUtils;
import com.elab.mq.consts.MqConstants;
import com.elab.mq.listener.ConsumerInterceptor;
import com.elab.mq.model.MessageModel;
import com.elab.mq.model.MqConsumerLogEntity;
import com.elab.mq.utils.RocketMonitorUtils;
import com.elab.spring.dao.IConsumerDao;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.net.InetAddress;
import java.util.Date;

/**
 * @Module 拦截器
 * @Description 消费者表拦截器
 * @Author liukaixiong
 * @Date 2020/12/30 15:16
 */
public class RMQMsgTableConsumerInterceptor implements ConsumerInterceptor {

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

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

    @Value(value = "${elab.mq.GROUP_ID:}")
    private String groupId;

    @Autowired
    private IConsumerDao consumerDao;

    @Override
    public boolean check(MessageModel messageModel,String groupName) {
        return checkRepeatMsg(messageModel);
    }

    @Override
    public void success(MessageModel messageModel,String groupName) {
        saveConsumerStatus(messageModel, MqConstants.MSG_OK, messageModel.getInvokeTime());
    }

    @Override
    public void error(MessageModel messageModel,String groupName, Throwable e) {
        saveConsumerStatus(messageModel, MqConstants.MSG_NO, messageModel.getInvokeTime());
        RocketMonitorUtils.sendConsumerMonitorData(messageModel, -1, e);
    }

    /**
     * 检查是否存在重复消费的消息
     *
     * @param message
     * @return
     * @throws Exception
     */
    protected boolean checkRepeatMsg(MessageModel message) {
        Integer producerId = 0;

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

        // 查询消息是否存在
        MqConsumerLogEntity consumerEntity = new MqConsumerLogEntity();
        consumerEntity.setProducerLogId(producerId);
        consumerEntity.setGroupId(groupId);
        consumerEntity.setModuleMethod(getClass().getName());
        consumerEntity.setInvokeStatus(MqConstants.MSG_OK);
        // 这里其实可以优化成只查询id是否存在
        Integer repeatCount = null;
        try {
            repeatCount = consumerDao.selectByObjectToCount(consumerEntity);
            if (repeatCount != null && repeatCount > 0) {
                logger.warn("消费者验证失败，已经消费成功过一次，不允许重复消费 已经成功的总数 : " + repeatCount + " 查询参数:" + JSON.toJSONString(consumerEntity));
                // throw new BusinessException("请不要重复消费，key：" + message);
                return false;
            }
            logger.debug("消费者验证通过，消费者消费的数据匹配上生产者数据，");
        } catch (Exception e) {
            logger.error("消息匹配异常", e);
        }
        return true;
    }

    /**
     * 消费完成之后的结果保存
     *
     * @param message     消息内容
     * @param status      状态
     * @param requestTime 请求时长
     */
    protected void saveConsumerStatus(MessageModel message, Integer status, long requestTime) {
        try {
            Integer producerId = DataUtils.getStringByIntegerValue(message.getProducerId());
            MqConsumerLogEntity consumerEntity = new MqConsumerLogEntity();
            consumerEntity.setProducerLogId(producerId);
            consumerEntity.setGroupId(groupId);
            consumerEntity.setModuleMethod(getClass().getName());
            consumerEntity.setInvokeStatus(status);
            consumerEntity.setCreatedTime(new Date());
            consumerEntity.setMsgId(message.getMsgID());
            consumerEntity.setMsgKey(message.getKey());
            consumerEntity.setMsgType(MqConstants.MQ_TYPE);
            consumerEntity.setApplicationIp(InetAddress.getLocalHost().getHostAddress());
            consumerEntity.setApplicationName(applicationName);
            consumerEntity.setLogId(Cat.getCurrentMessageId());
            consumerEntity.setPartitionName(message.getTag());
            consumerEntity.setTopic(message.getTopic());
            int insert = consumerDao.insert(consumerEntity);
            //logger.debug("插入kafka去重数据:" + insert);
        } catch (Exception e) {
            logger.error("消费记录日志添加失败", e);
        }
    }

}
