package com.education.business.message;

import com.education.business.message.param.QueueMessage;
import com.education.business.service.system.SystemMessageLogService;
import com.education.business.task.TaskManager;
import com.education.common.enums.MqMessageStatusEnum;
import com.education.model.entity.MessageLog;
import com.jfinal.json.Jackson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;

/**
 * mq 消息生产者工厂类
 * @author zjt
 * @create_at 2022年11月22日 0022 10:09
 * @since 1.0.6
 */
@Component
public class MqProducerFactory {

    private static Logger logger = LoggerFactory.getLogger(MqProducerFactory.class);
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private TaskManager taskManager;
    @Resource
    private SystemMessageLogService systemMessageLogService;

    private final Jackson jackson = new Jackson();


    /**
     * 异步发送mq队列消息
     */
    public void sendQueueMessage(QueueMessage queueMessage) {
        MessageLog exitsMessageLog = systemMessageLogService.selectByMessageId(queueMessage.getMessageId());
        if (exitsMessageLog != null) {
            return;
        }
        taskManager.pushTask(() -> {
            // 异步消息日志落库，提升系统性能
            MessageLog messageLog = new MessageLog();
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId(queueMessage.getMessageId());
            messageLog.setCorrelationDataId(queueMessage.getMessageId());
            messageLog.setCreateDate(new Date());
            messageLog.setExchange(queueMessage.getExchange());
            messageLog.setRoutingKey(queueMessage.getRoutingKey());
            String content = jackson.toJson(queueMessage);
            messageLog.setContent(content);
            try {
                // 发消息之前先进行消息落库
                systemMessageLogService.save(messageLog);
                rabbitTemplate.convertAndSend(queueMessage.getExchange(),
                        queueMessage.getRoutingKey(),
                        content, correlationData);
                logger.info("MQ发送消息:{}", content);
            } catch (Exception e) {
                logger.error("MQ消息发送异常: {}", content, e);
                messageLog.setStatus(MqMessageStatusEnum.SEND_FAIL.getCode());
                messageLog.setFailCause(e.getMessage());
                systemMessageLogService.updateById(messageLog);
            }
        });
    }
}
