package com.gkhy.safePlatform.safeCheck.mq.consumer;

import com.gkhy.safePlatform.commons.utils.JsonUtils;
import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckNoticeExecTaskMsg;
import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckWaitExecTaskMsg;
import com.gkhy.safePlatform.safeCheck.service.SafeCheckTaskUnitSchedulesService;
import com.gkhy.safePlatform.safeCheck.util.CalcuteDelayLevelUtils;
import com.gkhy.safePlatform.safeCheck.util.ConsolePrintIfEnvIsDevUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.Date;

@Service
@RocketMQMessageListener(topic = "${rocketmq.topic.safeCheckNoticeTaskTopic}",consumerGroup = "${rocketmq.consumers.safeCheckNoticeGroup}")
public class SafeCheckNoticeTaskConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

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

    @Value("${rocketmq.topic.safeCheckNoticeTaskTopic}")
    private String safeCheckNoticeTaskTopic;

    @Value("${rocketmq.topic.safeCheckWaitExecTopic}")
    private String safeCheckWaitExecTopic;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private SafeCheckTaskUnitSchedulesService taskUnitSchedulesService;

    @Override
    public void onMessage(MessageExt messageExt) {
        ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckNoticeTaskConsumer接收到消息  MSG ID : "+ messageExt.getMsgId()  + "  "+ new Date());
//        System.out.println("\n【##】SafeCheckNoticeTaskConsumer接收到消息\t MSG ID : "+ messageExt.getMsgId()  + "  "+ new Date());

        SafeCheckNoticeExecTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), SafeCheckNoticeExecTaskMsg.class);
        //logger.info("notice接收到消息内容" + msg.getTaskId() + msg.getNoticeTaskTime());

        if(msg!=null && msg.getTaskId()>0){
            //当前时间早于处理时间时
            Long relaySeconds = (msg.getNoticeTaskTime().getTime() - new Date().getTime())/1000;

            if(relaySeconds > 0){
                //重发消息，实现连续延时
                Message message = MessageBuilder.withPayload(msg).build();
                SendResult sendResult = rocketMQTemplate.syncSend(safeCheckNoticeTaskTopic, message,10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds));
                if(sendResult.getSendStatus() != SendStatus.SEND_OK){
                    throw new RuntimeException("【智能巡检】任务通知失败消息失败" + sendResult.getMsgId());
                }
                ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckNoticeTaskConsumer重发消息" + sendResult.getMsgId());
            }else {
                //TODO 通知消息暂无操作方法
                ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("notice暂时无动作");
                //通知完后开始计算执行时间
                SafeCheckWaitExecTaskMsg waitExecTaskMsg = new SafeCheckWaitExecTaskMsg();

                waitExecTaskMsg.setTaskId(msg.getTaskId());
                waitExecTaskMsg.setExecTaskTime(msg.getExecTaskTime());
                waitExecTaskMsg.setOutTaskTime(msg.getOutTaskTime());
                waitExecTaskMsg.setWorkId(msg.getWorkId());
                SendResult sendResult = rocketMQTemplate.syncSend(safeCheckWaitExecTopic, waitExecTaskMsg);
                ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】已经进行通知，发送消息至等待执行队列  TASK ID :"+msg.getTaskId() +" 时间："+ new Date());
                if(sendResult.getSendStatus() != SendStatus.SEND_OK){
                    throw new RuntimeException("【智能巡检】任务待巡检-->巡检倒计时失败"+msg.getTaskId());
                }
            }
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeThreadMin(4);
//        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
        try {
            consumer.subscribe(safeCheckNoticeTaskTopic,"*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}