package com.education.business.message.mq;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.education.business.message.param.QueueMessage;
import com.education.business.service.system.SystemMessageLogService;
import com.education.model.entity.MessageLog;
import com.jfinal.json.Jackson;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.core.ResolvableType;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import javax.annotation.Resource;

/**
 * mq消息监听器抽象类
 * @author zjt
 * @create_at 2022年11月21日 0021 17:12
 * @since 1.0.6
 */
public abstract class AbstractMqMessageListener<T extends QueueMessage> implements MqMessageListener<T> {

    private final Logger logger = LoggerFactory.getLogger(AbstractMqMessageListener.class);

    protected final Jackson jackson = new Jackson();

    @Resource
    private PlatformTransactionManager transactionManager;
    @Resource
    private TransactionDefinition transactionDefinition;
    @Resource
    private SystemMessageLogService systemMessageLogService;

    public void onMessage(Message message, Channel channel) {
        String content = new String(message.getBody());
        logger.info("收到MQ消息:{}", content);
        T queueMessage = jackson.parse(content, getTClass());
        // 验证消息是否已被消费(防止消息重复消费)
        String messageId = queueMessage.getMessageId();
        MessageLog messageLog = systemMessageLogService.selectByMessageId(messageId);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 校验消息是否已被消费成功, 已消费成功的mq消息将被自动删除 (防止业务消息已被处理，但是rabbitmq 手动确认消息失败导致消息重回消息队列)
        if (messageLog == null) {
            logger.warn("MQ消息:{}不存在或已被消费", content);
            basicAck(channel, deliveryTag, messageId);
            return;
        }
        TransactionStatus transactionStatus = transactionManager.getTransaction(transactionDefinition);
        try {
            doBizMessage(queueMessage);
            LambdaQueryWrapper queryWrapper = Wrappers.lambdaQuery(MessageLog.class)
                    .eq(MessageLog::getCorrelationDataId, messageId);
            // 消费成功删除mq消息日志记录
            systemMessageLogService.remove(queryWrapper);
            channel.basicAck(deliveryTag, false); // 手动确认消息 告诉rabbitmq 消息已消费
            transactionManager.commit(transactionStatus);
        } catch (Exception e) {
            transactionManager.rollback(transactionStatus);
            logger.error("MQ消息：{}消费异常", content, e);
        }
    }


    private void basicAck(Channel channel, long deliveryTag, String messageId) {
        try {
            channel.basicAck(deliveryTag, false); // 手动确认消息 告诉rabbitmq 消息已消费
        } catch (Exception e) {
            logger.error("mq消息:{}手动确认提交失败", messageId, e);
        }
    }


    public <T> Class<T> getTClass() {
        ResolvableType superType = ResolvableType.forClass(this.getClass()).getSuperType();
        ResolvableType[] types = superType.getGenerics();
        return  (Class<T>) types[0].resolve();
    }

    /**
     * 处理具体业务
     * @param value
     */
    abstract void doBizMessage(T value);
}
