package im.zksy.yjmqserver.system.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Component
public class MQTTSubsribe {
    private static Logger log = LoggerFactory.getLogger(MQTTSubsribe.class);
    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.host}")
    private String url;

    @Value("${spring.mqtt.qos}")
    private Integer qos;

    // 连接超时时间
    @Value("${spring.mqtt.completionTimeout}")
    private int completionTimeout;

    @Value("${spring.mqtt.topic}")
    private  String  topic;

    @Value("${spring.mqtt.clientId}")
    private  String  clientId;


    private String[] topics;

    private int[] qoslist;
    private MqttClient client;

    private MqttConnectOptions mqttConnectOptions;

    @Autowired
    private PushCallback pushCallback;

    private ScheduledExecutorService scheduled;

    public  void startReconnect() {
        this.scheduled = Executors.newSingleThreadScheduledExecutor();
        //    定时任务——重新连接mqtt服务器
        this.scheduled.scheduleAtFixedRate(new Runnable() {
            public void run() {
                if (!MQTTSubsribe.this.client.isConnected()) {
                    try {
                        log.info("---mqtt reconnect ---");
                        client.connect(mqttConnectOptions);
                        subscribe(Arrays.asList( topic.split(",")));
                    } catch (MqttSecurityException var2) {
                        var2.printStackTrace();
                    } catch (MqttException var3) {
                        var3.printStackTrace();
                    }
                }

            }
        }, 5000L, 10000L, TimeUnit.MILLISECONDS);
    }


    //	对mqttConnectOptions对象的常规设置
    public MqttConnectOptions getMqttConnectOptions() {
        this.mqttConnectOptions = new MqttConnectOptions();
        this.mqttConnectOptions.setCleanSession(true);
        if(!"".equals(username) ) {
            this.mqttConnectOptions.setUserName(username);
        }
        if(!"".equals(password) ) {
            this.mqttConnectOptions.setPassword(password.toCharArray());
        }
        this.mqttConnectOptions.setServerURIs(new String[]{url});
        this.mqttConnectOptions.setConnectionTimeout(completionTimeout);
        this.mqttConnectOptions.setKeepAliveInterval(2000);

        return mqttConnectOptions;
    }

    //	连接mqtt服务器订阅信息方法
//	topic也可作为参数传入
    public void subscribe(List<String> topics) {
        if(null!=topics &&topics.size()>0) {
            try {
                this.topics = topics.toArray(new String[topics.size()]);
                this.client = new MqttClient(url, clientId, new MemoryPersistence());
                this.getMqttConnectOptions();
                this.client.setCallback(this.pushCallback);
                this.client.connect(this.mqttConnectOptions);
                //遍历topic ，获得qs
                int[] qoslist=  new int[topics.size()];
                for( int i=0;i<topics.size();i++)
                {qoslist[i]=qos;
                }
                this.qoslist=qoslist;
//            可将订阅的一个或多个topic都存入数组中，同时订阅
//            String[] topic1 = {defaultTopic};
                this.client.subscribe(this.topics, qoslist);
                boolean connected = this.client.isConnected();
                log.info("连接状态为：" + connected);
                String flag = connected ? "成功" : "失败";
            } catch (MqttException e) {
                log.info(e.toString());
                startReconnect();
            }
        }else {
            log.info("topic is  empty");
        }
    }

    public void publishMessage(String topic, String message) {
        try {
            MqttMessage mqttmessage = new MqttMessage(message.getBytes());
            mqttmessage.setQos(this.qos);
            client.publish(topic, mqttmessage);
        } catch (MqttException me) {
            log.info(me.toString());
        }
    }

}