package com.njzscloud.common.mqtt.support;

import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.njzscloud.common.core.ex.Exceptions;
import com.njzscloud.common.core.jackson.Jackson;
import com.njzscloud.common.core.thread.ThreadPool;
import com.njzscloud.common.mqtt.config.MqttProperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.EventListener;

import java.lang.reflect.Method;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Consumer;

@Slf4j
public class MqttCliWrapper {
    static final Map<String, Consumer<MqttMsg>> fn = new ConcurrentHashMap<>();
    private final MqttClient client;
    private final ThreadPoolExecutor threadPool = ThreadPool.createThreadPool("MQTT 处理器");

    public MqttCliWrapper(MqttProperties mqttProperties) {
        log.info("mqtt 客户端已启用");
        String broker = mqttProperties.getBroker();
        String clientId = mqttProperties.getClientId() + "_" + new Date().getTime();
        String username = mqttProperties.getUsername();
        String password = mqttProperties.getPassword();
        String logDir = mqttProperties.getLogDir();
        try {
            MqttDefaultFilePersistence persistence = new MqttDefaultFilePersistence(logDir);
            client = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setAutomaticReconnect(true);
            options.setConnectionTimeout(5000);
            options.setMaxReconnectDelay(30000);
            if (StrUtil.isNotBlank(username) && StrUtil.isNotBlank(password)) {
                options.setUserName(username);
                options.setPassword(password.toCharArray());
            }
            client.connect(options);
            client.setCallback(new MsgHandler(fn, threadPool));
        } catch (Exception e) {
            throw Exceptions.error(e);
        }
    }

    public void shutdown() {
        try {
            if (client != null) {
                client.disconnect();
            }
            threadPool.shutdown();
        } catch (Exception e) {
            log.error("mqtt关闭失败", e);
        }
    }

    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos);
            log.info("mqtt 订阅消息：{} {}", topic, qos);
        } catch (Exception e) {
            log.error("mqtt 订阅失败：{} {}", topic, qos, e);
        }
    }

    public void subscribe(String topic) {
        this.subscribe(topic, 0);
    }

    public void subscribe(String topic, int qos, Consumer<MqttMsg> handler) {
        fn.put(topic, handler);
        this.subscribe(topic, qos);
    }

    public void subscribe(String topic, Consumer<MqttMsg> handler) {
        this.subscribe(topic, 0, handler);
    }

    public void publish(String topic, int qos, Object msg) {
        String jsonStr = null;
        try {
            jsonStr = Jackson.toJsonStr(msg);
            // log.info("mqtt 发布消息：{} {} {}", topic, qos, jsonStr);
            MqttMessage message = new MqttMessage(jsonStr.getBytes());
            message.setQos(qos);
            client.publish(topic, message);
        } catch (Exception e) {
            log.error("mqtt 发布消息失败：{} {} {}", topic, qos, jsonStr, e);
        }
    }

    public void publish(String topic, Object msg) {
        this.publish(topic, 0, msg);
    }

    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationReadyEvent() {
        ApplicationContext applicationContext = SpringUtil.getApplicationContext();
        String[] beanNames = applicationContext.getBeanDefinitionNames();
        for (String beanName : beanNames) {
            Object bean = applicationContext.getBean(beanName);
            process(bean);
        }
    }

    public void process(Object bean) throws BeansException {
        Class<?> clazz = bean.getClass();
        Method[] methods;
        boolean proxy = AopUtils.isCglibProxy(bean);
        if (proxy) {
            Class<?> parentClazz = clazz.getSuperclass();
            methods = parentClazz.getMethods();
        } else {
            methods = clazz.getMethods();
        }
        for (Method method : methods) {
            MqttListener mqttListener = method.getAnnotation(MqttListener.class);
            if (mqttListener == null) continue;
            if (proxy) {
                try {
                    String methodName = method.getName();
                    method = clazz.getMethod(methodName, method.getParameterTypes());
                } catch (NoSuchMethodException e) {
                    continue;
                }
            }
            String topic = mqttListener.topic();
            int qos = mqttListener.qos();
            this.subscribe(topic, qos);
            Consumer<MqttMsg> handler = new MsgConsumer(bean, method);
            fn.put(topic, handler);
        }
    }

    public void unsubscribe(String topic) {
        try {
            client.unsubscribe(topic);
        } catch (Exception e) {
            log.error("mqtt 取消订阅失败：{}", topic, e);
        }
    }


}
