|
|
@@ -0,0 +1,140 @@
|
|
|
+package mqttclient
|
|
|
+
|
|
|
+import (
|
|
|
+ "crypto/tls"
|
|
|
+ "errors"
|
|
|
+ "fmt"
|
|
|
+ mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
+ "log"
|
|
|
+ "time"
|
|
|
+)
|
|
|
+
|
|
|
+// MqttConnectConfig 连接的相关配置
|
|
|
+type MqttConnectConfig struct {
|
|
|
+ Broker string
|
|
|
+ User string
|
|
|
+ Password string
|
|
|
+ Certificate string //证书文件
|
|
|
+ PrivateKey string //秘钥
|
|
|
+ ClientId string
|
|
|
+ WillEnabled bool //遗愿
|
|
|
+ WillTopic string //遗愿主题
|
|
|
+ WillPayload string //遗愿消息
|
|
|
+ WillQos byte //遗愿服务质量
|
|
|
+ Qos byte //服务质量
|
|
|
+ Retained bool //保留消息
|
|
|
+ OnConnect mqtt.OnConnectHandler
|
|
|
+ OnConnectionLost mqtt.ConnectionLostHandler
|
|
|
+}
|
|
|
+
|
|
|
+// MqttClient MQTT客户端，此外也包含了几个参数
|
|
|
+type MqttClient struct {
|
|
|
+ qos byte
|
|
|
+ retained bool
|
|
|
+ Client mqtt.Client
|
|
|
+ topics map[string]mqtt.MessageHandler
|
|
|
+}
|
|
|
+
|
|
|
+// 新建证书，也可以不用
|
|
|
+func newTLSConfig(certFile string, privateKey string) (*tls.Config, error) {
|
|
|
+ cert, err := tls.LoadX509KeyPair(certFile, privateKey)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ return &tls.Config{
|
|
|
+ ClientAuth: tls.NoClientCert, //不需要证书
|
|
|
+ ClientCAs: nil, //不验证证书
|
|
|
+ InsecureSkipVerify: true, //接受服务器提供的任何证书和该证书中的任何主机名
|
|
|
+ Certificates: []tls.Certificate{cert},
|
|
|
+ }, nil
|
|
|
+}
|
|
|
+
|
|
|
+func NewMqttClient(config MqttConnectConfig) *MqttClient {
|
|
|
+ var c MqttClient
|
|
|
+ opts := mqtt.NewClientOptions().AddBroker(config.Broker).SetClientID(config.ClientId).SetMaxReconnectInterval(time.Second * 5)
|
|
|
+ if config.WillEnabled {
|
|
|
+ opts.SetWill(config.WillTopic, config.WillPayload, config.WillQos, config.Retained)
|
|
|
+ }
|
|
|
+ //判断是否设置证书
|
|
|
+ if config.Certificate != "" {
|
|
|
+ tlsConfig, err := newTLSConfig(config.Certificate, config.PrivateKey)
|
|
|
+ if err != nil {
|
|
|
+ log.Panic(err)
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ opts.SetTLSConfig(tlsConfig)
|
|
|
+ } else {
|
|
|
+ opts.SetUsername(config.User).SetPassword(config.Password)
|
|
|
+ }
|
|
|
+ //初始化
|
|
|
+ if config.OnConnect == nil {
|
|
|
+ config.OnConnect = func(c mqtt.Client) {}
|
|
|
+ }
|
|
|
+ if config.OnConnectionLost == nil {
|
|
|
+ config.OnConnectionLost = func(c mqtt.Client, err error) {}
|
|
|
+ }
|
|
|
+ opts.SetOnConnectHandler(c.connectHandler(config.OnConnect)).SetConnectionLostHandler(c.onConnectionLostHandler(config.OnConnectionLost))
|
|
|
+ c.Client = mqtt.NewClient(opts)
|
|
|
+ c.qos = config.Qos // qos的级别
|
|
|
+ c.retained = config.Retained // 保留消息
|
|
|
+ c.topics = make(map[string]mqtt.MessageHandler) //topic
|
|
|
+ // 用token的状态判断
|
|
|
+ if tc := c.Client.Connect(); tc.Wait() && tc.Error() != nil {
|
|
|
+ log.Panic(tc.Error())
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ return &c
|
|
|
+}
|
|
|
+
|
|
|
+// Publish Mqtt message.
|
|
|
+func (mc *MqttClient) Publish(topic string, payload []byte) error {
|
|
|
+ if mc != nil && mc.Client.IsConnected() {
|
|
|
+ if tc := mc.Client.Publish(topic, mc.qos, mc.retained, payload); tc.Wait() && tc.Error() != nil {
|
|
|
+ return tc.Error()
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ return errors.New("mqttClient is nil or disconnected")
|
|
|
+}
|
|
|
+
|
|
|
+// Subscribe subscribe a Mqtt topic.
|
|
|
+func (mc *MqttClient) Subscribe(topics []string, onMessage mqtt.MessageHandler) error {
|
|
|
+ for _, topic := range topics {
|
|
|
+ if tc := mc.Client.Subscribe(topic, mc.qos, onMessage); tc.Wait() && tc.Error() != nil {
|
|
|
+ return tc.Error()
|
|
|
+ }
|
|
|
+ mc.topics[topic] = onMessage
|
|
|
+ log.Println(fmt.Sprintf("订阅主题[%s]成功", topic))
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// Unsubscribe unsubscribe a Mqtt topic.
|
|
|
+func (mc *MqttClient) Unsubscribe(topics ...string) error {
|
|
|
+ if tc := mc.Client.Unsubscribe(topics...); tc.Wait() && tc.Error() != nil {
|
|
|
+ return tc.Error()
|
|
|
+ }
|
|
|
+ for _, topic := range topics {
|
|
|
+ delete(mc.topics, topic)
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+func (mc *MqttClient) Close() {
|
|
|
+ mc.Client.Disconnect(250) //Millisecond
|
|
|
+}
|
|
|
+
|
|
|
+func (mc *MqttClient) connectHandler(handler mqtt.OnConnectHandler) mqtt.OnConnectHandler {
|
|
|
+ return func(c mqtt.Client) {
|
|
|
+ for topic, onMessage := range mc.topics {
|
|
|
+ mc.Client.Subscribe(topic, mc.qos, onMessage)
|
|
|
+ }
|
|
|
+ handler(c)
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (mc *MqttClient) onConnectionLostHandler(handler mqtt.ConnectionLostHandler) mqtt.ConnectionLostHandler {
|
|
|
+ return func(c mqtt.Client, e error) {
|
|
|
+ handler(c, e)
|
|
|
+ }
|
|
|
+}
|
houyaf