package com.muyu.parsing.mqtt.service;

import com.muyu.cargateway.domain.properties.MqttProperties;
import com.muyu.parsing.manager.MessageProcessor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class MqttClientService {

    @Autowired
    private MessageProcessor messageProcessor;

    private final ExecutorService executorService = Executors.newCachedThreadPool();
    private MqttClient sampleClient;


    public void connectAndSubscribeAsync(MqttProperties mqttProperties) {
        executorService.submit(() -> {
            try {
                connectAndSubscribe(mqttProperties);
            } catch (MqttException | IOException e) {
                log.error("MQTT连接或订阅失败", e);
            }
        });
    }

    private void connectAndSubscribe(MqttProperties mqttProperties) throws MqttException, IOException {
//        if (sampleClient != null && sampleClient.isConnected()) {
//            log.info("MQTT客户端已经连接，跳过重新连接。");
//            return;
//        }

        sampleClient = new MqttClient(mqttProperties.getBroker(), mqttProperties.getClientId());
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(mqttProperties.getUserName());
        connOpts.setPassword(mqttProperties.getPassword().toCharArray());
        connOpts.setCleanSession(true);

        sampleClient.connect(connOpts);
        sampleClient.subscribe(mqttProperties.getTopic(), 0);

        sampleClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                log.error("连接丢失：{}", throwable.getMessage());
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) {
                executorService.submit(() -> messageProcessor.processMessage(mqttMessage));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                log.info("消息发送完成：{}", iMqttDeliveryToken);
            }
        });
    }

    @PreDestroy
    public void shutdown() {
        executorService.shutdown();
        if (sampleClient != null && sampleClient.isConnected()) {
            try {
                sampleClient.disconnect();
            } catch (MqttException e) {
                log.error("MQTT客户端断开连接失败", e);
            }
        }
    }
}
