package com.jianshui.queue.listener;

import com.alibaba.fastjson.JSONObject;
import com.jianshui.common.utils.RedisUtil;
import com.jianshui.queue.consumer.RedisQueueConsumer;
import com.jianshui.queue.utils.RedisQueueUtil;
import com.jianshui.queue.dto.RedisQueueMessage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @Description
 * @Author 巩权林
 * @Date 2021-11-25 9:58
 **/
public class RedisQueueListener implements Runnable {

    private final Long WAITTIME = 30L;

    private RedisQueueConsumer redisQueueConsumer;

    protected final Logger log = LoggerFactory.getLogger(RedisQueueListener.class);


    public RedisQueueListener(RedisQueueConsumer redisQueueConsumer) {
        this.redisQueueConsumer = redisQueueConsumer;
    }

    @Override
    public void run() {

        try {
            log.info("【redis队列】 监听器开始监听:{}", redisQueueConsumer.getQueueName());
            // while (RedisQueueConsumerContainer.isRun) {
            while (true) {
                RedisQueueMessage msg = new RedisQueueMessage(null);
                try {
                    log.info("【redis队列】进入消费 {}", redisQueueConsumer.getQueueName());
                    Jedis jedis = RedisUtil.getJedis();
                    if (jedis == null) {
                        log.info("【redis队列】获取jedis失败，jedis为空！");
                        Thread.sleep(500);
                        continue;
                    }
                    log.info("【redis队列】获取到redis线程 {}", redisQueueConsumer.getQueueName());
                    // 阻塞的读取消息一把，每次阻塞30秒，防止空转，阻塞redis
                    List<String> rawMsg = jedis.brpop(0, redisQueueConsumer.getQueueName());
                    if (rawMsg != null && rawMsg.size() > 0) {
                        msg = JSONObject.parseObject(rawMsg.get(1), RedisQueueMessage.class);
                        log.info("【redis队列】jedis读取到队列消息，内容为{}", JSONObject.toJSONString(msg));
                        try {
                            redisQueueConsumer.consume(msg);
                        } catch (Exception e) {
                            log.info("【redis队列】消费异常 {}", redisQueueConsumer.getQueueName());
                            e.printStackTrace();
                        }
                    }
                    log.info("【redis队列】队列消费完成{}", redisQueueConsumer.getQueueName());
                    // String rawMsg = jedis.rpop(redisQueueConsumer.getQueueName());
                    // msg = JSONObject.parseObject(rawMsg, RedisQueueMessage.class);
                    // if (msg != null) {
                    //     redisQueueConsumer.consume(msg);
                    // }
                    jedis.close();
                    log.info("【redis队列】当前jedis关闭{}", redisQueueConsumer.getQueueName());
                    Thread.sleep(500);
                    log.info("【redis队列】进入循环{}", redisQueueConsumer.getQueueName());
                } catch (Exception e) {
                    log.error("【redis队列】redis监听器错误：{}", redisQueueConsumer.getQueueName());
                    e.printStackTrace();
                    if (msg.getRetry() != null && msg.getRetry() > 0) {
                        RedisQueueMessage retryMsg = JSONObject.parseObject(JSONObject.toJSONString(msg), RedisQueueMessage.class);
                        Integer retryNum = retryMsg.getRetry();
                        if (retryNum != null && retryNum > 0) {
                            retryNum = retryNum - 1;
                            retryMsg.setRetry(retryNum);
                        }
//                    RedisQueueUtil.build().onQueue("check_consumer")
                        RedisQueueUtil.reBuild(retryMsg).dispatch();
                        log.error("【redis队列】redis监听器错误，重新进入队列排队：{}", redisQueueConsumer.getQueueName());
                    } else {
                        // 如果都失败了，走onError处理
                        redisQueueConsumer.error(msg);
                        log.error("【redis队列】redis监听器错误：{}", redisQueueConsumer.getQueueName());
                    }
                }
            }


        } catch (Exception e) {
            log.info("【redis队列】 监听器出现异常，重新投递:{}", redisQueueConsumer.getQueueName());
            e.printStackTrace();
            this.run();
        }
        log.error("【redis队列】 监听器出现退出，重新投递:{}", redisQueueConsumer.getQueueName());


    }
}
