Cold_mqtt/tests/device/MqttServer.go

package MqttServer

import (
	"fmt"
	"github.com/yosssi/gmq/mqtt"
	"github.com/yosssi/gmq/mqtt/client"
	"time"
)

var (
	cli    *client.Client
	T_sn   string
	T_pass string
)

func init() {
	T_sn = "2024168563127390"
	T_pass = "qMua4PZyl2vwchNL"
	Run_MqttServer()
}

/*
{
	"type": 203,
	"data": {
		"sn": "2024666666666666",
		"mqtt1": "mqttyuht.coldbaozhida.com",
		"mqtt2": "mqttlodr.coldbaozhida.com",
		"mqtt3": "mqttyuht.coldbaozhida.com",
		"mqtt4": "",
		"mqtt5": "",
		"port": 1883,
		"user": "2024666666666666",
		"pass": "qMua4PZyl2vwchNL"
	}
}
*/

func Run_MqttServer() {
	time.Sleep(3 * time.Second)
	fmt.Println("============Run_MqttServer=============", "")

	// Create an MQTT Client.
	cli = client.New(&client.Options{
		// Define the processing of the error handler.
		ErrorHandler: func(err error) {
			fmt.Println("err!!!!!! Run_MqttServer:", err.Error())

			time.Sleep(3 * time.Second)
			go Run_MqttServer() // MQTT 通讯
			return
		},
	})

	// Terminate the Client.
	defer cli.Terminate()

	c := client.ConnectOptions{
		Network:  "tcp",
		Address:  "mqttyuht.coldbaozhida.com",
		ClientID: []byte(T_sn + "_test"),
		UserName: []byte(T_sn),
		Password: []byte(T_pass),
	}
	fmt.Println("Address：", c.Address)
	fmt.Println("ClientID：", string(c.ClientID))

	// Connect to the MQTT Server.
	err := cli.Connect(&c)

	if err != nil {
		fmt.Println("MqttServer", "连接MQTT失败 [cli.Connect]", "")
		fmt.Println("err!!!!!! Run_MqttServer:", "连接MQTT失败：", err)
		fmt.Println("err!!!!!! 连接MQTT失败:", err)
		cli.Terminate()
		time.Sleep(3 * time.Second)
		go Run_MqttServer() // MQTT 通讯

		return
	}

	// Subscribe to topics.
	err = cli.Subscribe(&client.SubscribeOptions{
		SubReqs: []*client.SubReq{
			&client.SubReq{
				TopicFilter: []byte("/pub/2024168563127390"),
				QoS:         mqtt.QoS0,
				// Define the processing of the message handler.
				Handler: func(topicName, message []byte) {
					fmt.Println(">-" + string(topicName) + " " + string(message))

				},
			},
		},
	})
	if err != nil {
		fmt.Println("MqttServer", "订阅消息 [cli.Subscribe]", "")
		fmt.Println("err!!!!!! Run_MqttServer:", "连接MQTT失败：", err)

	}

	fmt.Println("MQTT ok!")
}

// 发送数据
func Mqtt_publish(topic string, text string) {

	// Publish a message.
	err := cli.Publish(&client.PublishOptions{
		QoS:       mqtt.QoS0,
		TopicName: []byte("/pub/" + topic),
		Message:   []byte(text),
	})

	fmt.Println("-> /pub/" + topic + " " + text)
	if err != nil {
		fmt.Println("MqttServer", "发送消息失败 [cli.Publish]", text)
	}

}