package com.inspect.analysis.service.impl;

import com.alibaba.nacos.shaded.com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;

import java.time.Duration;
import java.util.function.Consumer;

public class RedisQueueConsumer<T> {
    private static final Logger logger = LoggerFactory.getLogger(RedisQueueConsumer.class);

    private final RedisTemplate<String, String> redisTemplate;
    private final Gson gson = new Gson();
    private final String queueKey;
    private final Class<T> clazz;

    private final long intervalMs;
    private final long idleSleepMs;

    private final Consumer<T> handler;

    public RedisQueueConsumer(RedisTemplate<String, String> redisTemplate,
                              String queueKey,
                              Class<T> clazz,
                              long intervalMs,
                              long idleSleepMs,
                              Consumer<T> handler) {
        this.redisTemplate = redisTemplate;
        this.queueKey = queueKey;
        this.clazz = clazz;
        this.intervalMs = intervalMs;
        this.idleSleepMs = idleSleepMs;
        this.handler = handler;
    }

    public void consumeLoop() {
        while (true) {
            try {
                // 阻塞式弹出元素（避免忙等）
                String json = redisTemplate.opsForList()
                        .leftPop(queueKey, Duration.ofSeconds(1));

                if (json == null) {
                    Thread.sleep(idleSleepMs);
                    break; // 本轮结束，等待下次 @Scheduled 调度
                }

                long start = System.currentTimeMillis();

                T obj = gson.fromJson(json, clazz);
                handler.accept(obj); // 业务处理回调

                // 节流处理
                long cost = System.currentTimeMillis() - start;
                if (cost < intervalMs) {
                    Thread.sleep(intervalMs - cost);
                }

            } catch (Exception e) {
                logger.error("RedisQueueConsumer error", e);
            }
        }
    }
}
