92 lines
1.9 KiB
Go
92 lines
1.9 KiB
Go
package mqtt
|
|
|
|
import (
|
|
"fmt"
|
|
"log"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
)
|
|
|
|
type MqttClient struct {
|
|
Client mqtt.Client
|
|
SubscribeList []SubscribeConf
|
|
}
|
|
|
|
type MqttConf struct {
|
|
Host string
|
|
Port string
|
|
Username string
|
|
Password string
|
|
ClientId string
|
|
}
|
|
|
|
type SubscribeConf struct {
|
|
Topic string
|
|
Handler func(mqtt.Client, mqtt.Message)
|
|
}
|
|
|
|
func NewMqttClient() *MqttClient {
|
|
c := &MqttClient{}
|
|
c.Client = nil
|
|
c.SubscribeList = make([]SubscribeConf, 0)
|
|
|
|
return c
|
|
}
|
|
|
|
func (c *MqttClient) SubTopic(topic string, f func(mqtt.Client, mqtt.Message)) {
|
|
c.SubscribeList = append(c.SubscribeList, SubscribeConf{
|
|
Topic: topic,
|
|
Handler: f,
|
|
})
|
|
}
|
|
|
|
func (c *MqttClient) Connect(conf MqttConf) {
|
|
host := conf.Host
|
|
port := conf.Port
|
|
username := conf.Username
|
|
password := conf.Password
|
|
clientId := conf.ClientId
|
|
|
|
opts := mqtt.NewClientOptions()
|
|
opts.AddBroker(fmt.Sprintf("tcp://%s:%s", host, port))
|
|
opts.SetAutoReconnect(true) // 自动重连
|
|
opts.SetUsername(username)
|
|
opts.SetPassword(password)
|
|
opts.SetClientID(clientId)
|
|
|
|
opts.SetOnConnectHandler(func(client mqtt.Client) {
|
|
log.Println("mqtt 连接成功")
|
|
|
|
// 创建订阅
|
|
log.Println("加载订阅...")
|
|
c.CreateSubscribe()
|
|
log.Println("加载订阅完成")
|
|
})
|
|
|
|
opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
|
|
log.Println("mqtt 连接断开")
|
|
})
|
|
|
|
client := mqtt.NewClient(opts)
|
|
|
|
if token := client.Connect(); token.Wait() && token.Error() != nil {
|
|
panic("创建mqtt客户端错误,错误信息: " + token.Error().Error())
|
|
}
|
|
|
|
c.Client = client
|
|
}
|
|
|
|
func (c *MqttClient) Publish(topic string, payload []byte) {
|
|
token := c.Client.Publish(topic, 0, false, payload)
|
|
token.Wait()
|
|
}
|
|
|
|
func (c *MqttClient) CreateSubscribe() {
|
|
for _, conf := range c.SubscribeList {
|
|
if token := c.Client.Subscribe(conf.Topic, 1, conf.Handler); token.Wait() && token.Error() != nil {
|
|
log.Fatal(token.Error())
|
|
}
|
|
log.Printf("订阅 %s 完成\n", conf.Topic)
|
|
}
|
|
}
|