package com.huashe.park.collect.core;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.alibaba.fastjson2.JSONObject;
import com.huashe.common.utils.StringUtils;

public abstract class TopicMsgEngine implements ITopicListener, MqttMessageHandler {
    private final static Logger logger = LoggerFactory.getLogger(TopicMsgEngine.class);

    private String engineName;

    @Value("${evt-fusion.thread-pool-size:2}")
    private int threadPoolSize;

    @Resource(name = "bdThreadPoolTaskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    public void init() {
        List<LinkedBlockingQueue<JSONObject>> queue = getQueue();
        for (int i = 0; i < threadPoolSize; i++) {
            queue.add(new LinkedBlockingQueue<>());
        }
    }

    public void setEngineName(String name) {
        this.engineName = name;
    }

    public void push(JSONObject msg) {
        String key = msg.getString("key");
        if (StringUtils.isEmpty(key)) {
            return;
        }
        int bucketIndex = computeHashModulo(key, threadPoolSize);
        List<LinkedBlockingQueue<JSONObject>> queue = getQueue();
        LinkedBlockingQueue<JSONObject> messageQueue = queue.get(bucketIndex);
        messageQueue.offer(msg);
    }

    public void start() {
        List<LinkedBlockingQueue<JSONObject>> queue = getQueue();
        for (int i = 0; i < threadPoolSize; i++) {
            int finalI = i;
            threadPoolTaskExecutor.execute(() -> {
                while (true) {
                    try {
                        JSONObject msg = queue.get(finalI).take();
                        messageBizHandle(msg);
                    }
                    catch (Exception e) {
                        logger.error("{} error", this.engineName, e);
                    }
                }
            });
        }
    }

    /**
     * 计算字符串的哈希值并对某个数取余数。
     *
     * @param str 字符串
     * @param modulo 取余数的基数
     * @return 字符串哈希值对 modulo 取余的结果
     */
    private int computeHashModulo(String str, int modulo) {
        long hash = 0;
        long base = 31;
        long baseMod = base % modulo; // 预计算乘法因子

        for (char c : str.toCharArray()) {
            hash = (hash * baseMod + c) % modulo; // 避免整数溢出
        }

        return (int) (hash % modulo); // 最终结果转换为 int
    }

}
