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

import com.gkhy.safePlatform.commons.utils.JsonUtils;
import com.gkhy.safePlatform.safeCheck.entity.SafeCheckTask;
import com.gkhy.safePlatform.safeCheck.entity.SafeCheckWork;
import com.gkhy.safePlatform.safeCheck.enums.TaskStatusEnum;
import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckTimeOutTaskMsg;
import com.gkhy.safePlatform.safeCheck.service.SafeCheckTaskUnitSchedulesService;
import com.gkhy.safePlatform.safeCheck.service.baseService.SafeCheckWorkService;
import com.gkhy.safePlatform.safeCheck.util.CalcuteDelayLevelUtils;
import com.gkhy.safePlatform.safeCheck.util.ConsolePrintIfEnvIsDevUtil;
import org.apache.commons.lang3.ObjectUtils;
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.safeCheckTimeOutTaskTopic}",consumerGroup = "${rocketmq.consumers.safeCheckTimeOutGroup}")
public class SafeCheckTimeOutTaskConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

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


    @Autowired
    private RocketMQTemplate rocketMQTemplate;

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

    @Autowired
    private SafeCheckWorkService safeCheckWorkService;

    @Autowired
    private SafeCheckTaskUnitSchedulesService safeCheckTaskUnitSchedulesService;

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


        SafeCheckTimeOutTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), SafeCheckTimeOutTaskMsg.class);
        //logger.info("TimeOutTaskConsumer接收到消息内容" + msg.getTaskId()+msg.getWorkId()+msg.getOutTaskTime());
        //TODO 要对过期的消息做处理，可能要使用配置中心
        if(msg!=null && msg.getWorkId()>0){
            //当前时间早于处理时间时
            Long relaySeconds = (msg.getOutTaskTime().getTime() - new Date().getTime())/1000;
            if(relaySeconds > 0){
                //重发消息，实现连续延时
                Message message = MessageBuilder.withPayload(msg).build();
                SendResult sendResult = rocketMQTemplate.syncSend(safeCheckTimeOutTaskTopic, message, 10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds));
                //logger.info("重发内容=  "+ message + " " + message.getPayload() + " " +relaySeconds+ " "  + CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds));
                if(sendResult.getSendStatus() != SendStatus.SEND_OK){
                    throw new RuntimeException("【智能巡检】任务到期倒计时通知消息失败" + msg.getOutTaskTime());
                }
                ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【智能巡检】SafeCheckTimeOut重发消息" + sendResult.getMsgId());
            }else {
                SafeCheckTask task = safeCheckTaskUnitSchedulesService.getTaskById(msg.getTaskId());

                if (ObjectUtils.isEmpty(task.getResultStatus())){
                    //设置任务超时
                    ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【智能巡检】SafeCheckTimeOut设置任务状态为超时...");
                    if (safeCheckTaskUnitSchedulesService.updateTaskStatus(msg.getTaskId(), TaskStatusEnum.TASK_TIMEOUT_NO_CHECK.getStatus()) <= 0){
                        throw new RuntimeException("【智能巡检】任务状态设置超时失败" + msg.getOutTaskTime());
                    }
                }
                Date execTaskTime = msg.getExecTaskTime();
                SafeCheckWork workById = safeCheckWorkService.getWorkById(msg.getWorkId());
                ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【智能巡检】任务开始执行时在调度信息中记录的开始时间"+execTaskTime+"----->>任务结束后调度信息中的时间:"+workById.getNextCheckTime());
                //说明这次调度时间没有因为中断 导致调度时间被修改了
                if (workById.getNextCheckTime().compareTo(execTaskTime) == 0) {
                    //设置作业为开启
                    ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【智能巡检】SafeCheckTimeOut重置作业状态开始...");
                    if (safeCheckTaskUnitSchedulesService.resetWorkStatus(msg.getWorkId()) <= 0) {
                        throw new RuntimeException("【智能巡检】作业状态重置为开启状态失败" + msg.getOutTaskTime());
                    }
                }
                ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【智能巡检】SafeCheckTimeOut重置作业状态完成,等待下一次调度...");
            }
        }
    }

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