package com.njzscloud.common.mqtt.support;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Consumer;

@Slf4j
public class MsgHandler implements MqttCallback {
    private final Map<String, Consumer<MqttMsg>> fn;
    private final ThreadPoolExecutor threadPool;

    public MsgHandler(Map<String, Consumer<MqttMsg>> fn, ThreadPoolExecutor threadPool) {
        this.threadPool = threadPool;
        this.fn = fn;
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
        Consumer<MqttMsg> handler = fn.get(topic);
        if (handler == null) return;
        try {
            CompletableFuture
                .runAsync(() -> handler.accept(new MqttMsg(message.getPayload())), threadPool)
                .exceptionally(e -> {
                    log.error("mqtt 消息处理失败：{} {}", topic, new String(message.getPayload()), e);
                    return null;
                });
        } catch (Exception e) {
            log.error("mqtt 消息处理失败", e);
        }
    }

    public void connectionLost(Throwable cause) {
        log.error("mqtt 连接已断开", cause);
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        // log.info("消息投递结果：{}", token.isComplete());
    }
}
