package com.elab.mq.msg.impl;

import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
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.core.utils.StringUtils;
import com.elab.mq.dao.IProducerDao;
import com.elab.mq.model.MessageModel;
import com.elab.mq.model.ProducerEntity;
import com.elab.mq.model.SendResultModel;
import com.elab.mq.msg.IMsgProducerFacade;
import com.elab.mq.msg.adptor.SendCallbackAdaptor;
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;
import java.util.Properties;

/**
 * 消息生产包装类
 *
 * @author Liukx
 * @create 2018-04-09 18:23
 * @email liukx@elab-plus.com
 **/
public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade {


    private Logger logger = LoggerFactory.getLogger(MsgProducerImpl.class);

    @Autowired
    private IProducerDao producerDao;

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

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

    private final String MQ_PRODUCER = "MQ_PRODUCER";


    /**
     * 往生产者表中写入数据
     *
     * @param message
     * @return
     */
    private int insertRecordProducer(MessageModel message) {
        String content = ObjectUtils.objectParseJsonStr(message.getObject(Object.class));
        ProducerEntity producerEntity = new ProducerEntity();
        String tag = message.getTag();
        if (StringUtils.isEmpty(message.getModuleName())) {
            producerEntity.setModuleName(moduleName);
        } else {
            producerEntity.setModuleName(message.getModuleName());
        }
        producerEntity.setTag(tag);
        producerEntity.setCatId(Cat.getCurrentMessageId());
        producerEntity.setGroupId(groupId);
        producerEntity.setHouseId(message.getHouseId());
        producerEntity.setProducerStatus(0);
        producerEntity.setGroupId(message.getGroupId());
        producerEntity.setTopicId(message.getTopic());
        producerEntity.setContent(content);
        producerEntity.setStatus(1);
        producerEntity.setCreated(DateUtils.getCurrentDateTime());
        int id = 0;
        try {
            id = producerDao.insert(producerEntity);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("往生产者表中插入记录失败：，入参：" + ObjectUtils.objectParseJsonStr(message), e);
        }
        return id;
    }

    /**
     * 修改生产者数据
     *
     * @param id
     * @param result
     * @return
     */
    private void updateRecordProducer(int id, Integer status, String result) {
        try {
            ProducerEntity producerEntity = producerDao.selectById(id + "");
            if (producerEntity != null) {
                producerEntity.setProducerStatus(status);
                producerEntity.setResultContent(result);
                producerDao.updateById(producerEntity);
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("修改生产者表中记录失败：，入参：" + id, e);
        }
    }

    @Override
    public void sendAsync(MessageModel message, SendCallbackAdaptor sendCallback) {
        logger.debug(" 发送一条异步消息 " + message + " -> " + message.toString() + " 回调类 : " + sendCallback.getClass().getName());
        super.sendAsync(message, sendCallback);
    }

    @Override
    public SendResultModel send(MessageModel message) {
        String topic = message.getTopic();
        String tag = message.getTag();
        Transaction t = Cat.newTransaction(MQ_PRODUCER, topic + "_" + tag + "_" + getClass().getSimpleName());
        logger.debug(" 发送一条消息 " + message.getMsgID() + " -> " + message.toString());
        int id = insertRecordProducer(message);
        try {
            message.setProducerId(id + "");
            SendResult result = super.send(message);
            updateRecordProducer(id, 1, result.toString());
            logger.debug(" 消息发送结果 : " + result.toString());
            t.setSuccessStatus();
            return new SendResultModel(result);
        } catch (Exception e) {
            logger.error("消息发送异常", e);
            t.setStatus(e);
            updateRecordProducer(id, -1, e.getMessage());
        } finally {
            t.complete();
        }
        return null;
    }

    @Override
    public void sendOneway(MessageModel message) {
        logger.debug(" 发送一条单向消息 " + message.getMsgID() + " -> " + message.toString());
        super.sendOneway(message);
    }

    @Override
    public SendResultModel sendDeliverTime(MessageModel messageModel, int seconds) {
        logger.debug(" 发送一条延时消息 : 将在[" + seconds + "]秒后发送 , 消息内容 : " + messageModel.toString());
        long time = System.currentTimeMillis() + (seconds * 1000);
        messageModel.setStartDeliverTime(time);
        return send(messageModel);
    }

    @Override
    public SendResultModel sendTimingMsg(MessageModel messageModel, Date datetime) {
        logger.debug(" 发送一条定时消息 : 将在[" + datetime + "]后发送 , 消息内容 : " + messageModel.toString());
        messageModel.setStartDeliverTime(datetime.getTime());
        return send(messageModel);
    }

    @Override
    public void setProperties(Properties properties) {
        super.setProperties(properties);
    }
}
