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

import com.gkhy.safePlatform.commons.utils.CalcuteDelayLevelUtils;
import com.gkhy.safePlatform.doublePrevention.entity.PreventDangerCheckTask;
import com.gkhy.safePlatform.doublePrevention.mq.msg.PreventWaitExecTaskMsg;
import com.gkhy.safePlatform.doublePrevention.service.DangerService;
import com.gkhy.safePlatform.doublePrevention.service.baseService.PreventDangerCheckTaskService;
import com.gkhy.safePlatform.commons.utils.JsonUtils;
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.redisson.api.RLock;
import org.redisson.api.RedissonClient;
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;
import java.util.concurrent.TimeUnit;

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

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

    @Value("${rocketmq.topic.preventWaitWorkTopic}")
    private String preventWaitWorkTopic;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private DangerService dangerService;
    @Autowired
    private PreventDangerCheckTaskService preventDangerCheckTaskService;

    @Override
    public void onMessage(MessageExt messageExt) {
        System.out.println("\n【双重预防】preventWaitExecConsumer接收到消息\t MSG ID : "+messageExt.getMsgId()  + new Date());

        PreventWaitExecTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), PreventWaitExecTaskMsg.class);

        //logger.info("WaitExecConsumer接收到消息内容" + msg.getTaskId() + msg.getExecTaskTime());
        if(msg!=null && msg.getTaskId()>0){
            //当前时间早于处理时间时
            Long relaySeconds = (msg.getExecTaskTime().getTime() - new Date().getTime())/1000;
            if(relaySeconds > 0){
                //重发消息，实现连续延时
                Message message = MessageBuilder.withPayload(msg).build();
                SendResult sendResult = rocketMQTemplate.syncSend(preventWaitWorkTopic, message,10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds));
                if(sendResult.getSendStatus() != SendStatus.SEND_OK){
                    throw new RuntimeException("任务通知失败消息失败" + msg.getExecTaskTime());
                }
                logger.info("\n【双重预防】preventWaitExecConsumer消息重发" + sendResult.getMsgId());
             //   logger.info("\nWaitExecConsumer消息重发" + message);
            }else {
                //1-1、加分布式锁-锁workId
                String lockName = "PREVENT_WAIT_EXEC_" + msg.getTaskId();
                RLock waitExecLock = redissonClient.getLock(lockName);
                waitExecLock.lock(3, TimeUnit.SECONDS);
                //1-2检查加锁的作业状态是否已经修改
                //如果任务为执行中，结束方法
                PreventDangerCheckTask taskById = preventDangerCheckTaskService.getTaskById(msg.getTaskId());
                if (taskById.getTaskStatus() == 4){
                    return;
                }
                //设置任务为执行中
                logger.info("\n【双重预防】preventWait设置任务状态为执行中");
                Byte taskStatus = 4;
                if (preventDangerCheckTaskService.updateTaskStatus(msg.getTaskId(),taskStatus) <0){
                    throw new RuntimeException("任务状态更新失败" + msg.getExecTaskTime());
                }
                //释放分布式锁
                waitExecLock.unlock();
            }
        }else {
            throw new RuntimeException("preventWaitExecConsumer消费失败" + msg.getExecTaskTime());
        }
        //System.out.println("[MQ] 接收到消息:\t"+new String(messageExt.getBody()));
    }

    @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(preventWaitWorkTopic,"*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}