init

.gitignore

@@ -0,0 +1 @@
+/.idea/

client_instance.go

@@ -0,0 +1,26 @@
+package mqtt_client
+
+var clientInstance *Client
+
+func InitInstance(opts *ClientOptions) {
+	if clientInstance == nil {
+		client, err := New(opts)
+		if err != nil {
+			panic(err)
+		}
+
+		clientInstance = client
+	}
+}
+
+func DestroyInstance() {
+	if clientInstance == nil {
+		return
+	}
+
+	Destroy(clientInstance)
+}
+
+func GetInstance() *Client {
+	return clientInstance
+}

go.mod

@@ -0,0 +1,11 @@
+module git.sxidc.com/go-tools/mqtt_client
+
+go 1.21.3
+
+require github.com/eclipse/paho.mqtt.golang v1.4.3
+
+require (
+	github.com/gorilla/websocket v1.5.0 // indirect
+	golang.org/x/net v0.8.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
+)

go.sum

@@ -0,0 +1,8 @@
+github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik=
+github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
+golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

mqtt_client.go

@@ -0,0 +1,125 @@
+package mqtt_client
+
+import (
+	"errors"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"time"
+)
+
+type MessageHandler func(client *Client, topic string, data []byte)
+type OnConnectHandler func(client *Client)
+type ConnectionLostHandler func(client *Client, err error)
+
+type Client struct {
+	client mqtt.Client
+}
+
+type ClientOptions struct {
+	UserName              string
+	Password              string
+	Address               string
+	ClientID              string
+	KeepAliveSec          time.Duration
+	PingTimeoutSec        time.Duration
+	OnConnectHandler      OnConnectHandler
+	ConnectionLostHandler ConnectionLostHandler
+}
+
+func (opt *ClientOptions) check() error {
+	if isStringEmpty(opt.UserName) {
+		return errors.New("必须传递用户名")
+	}
+
+	if isStringEmpty(opt.Password) {
+		return errors.New("必须传递密码")
+	}
+
+	if isStringEmpty(opt.Address) {
+		return errors.New("必须传递地址")
+	}
+
+	if isStringEmpty(opt.ClientID) {
+		return errors.New("必须传递客户端ID")
+	}
+
+	if opt.OnConnectHandler == nil {
+		return errors.New("必须传递连接回调")
+	}
+
+	return nil
+}
+
+func New(opts *ClientOptions) (*Client, error) {
+	if opts == nil {
+		return nil, errors.New("必须传递参数")
+	}
+
+	client := new(Client)
+
+	mqttClient := mqtt.NewClient(mqtt.NewClientOptions().
+		SetAutoReconnect(true).
+		SetUsername(opts.UserName).
+		SetPassword(opts.Password).
+		AddBroker(opts.Address).
+		SetClientID(opts.ClientID).
+		SetKeepAlive(opts.KeepAliveSec).
+		SetPingTimeout(opts.PingTimeoutSec).
+		SetWill(opts.ClientID+"/will", "dead", 2, true).
+		SetOnConnectHandler(func(mqttClient mqtt.Client) {
+			opts.OnConnectHandler(client)
+		}).
+		SetConnectionLostHandler(func(mqttClient mqtt.Client, err error) {
+			if opts.ConnectionLostHandler != nil {
+				opts.ConnectionLostHandler(client, err)
+			}
+		}))
+
+	client.client = mqttClient
+
+	return client, nil
+}
+
+func Destroy(client *Client) {
+	client.client.Disconnect(250)
+	client = nil
+}
+
+func (client *Client) Connect() error {
+	token := client.client.Connect()
+	token.Wait()
+	err := token.Error()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (client *Client) Disconnect(quiesce uint) {
+	client.client.Disconnect(quiesce)
+}
+
+func (client *Client) Publish(topic string, qos byte, retained bool, payload string) error {
+	token := client.client.Publish(topic, qos, retained, payload)
+	token.Wait()
+	return token.Error()
+}
+
+func (client *Client) Subscribe(topic string, qos byte, handlerFunc MessageHandler) error {
+	if handlerFunc == nil {
+		return errors.New("必须传递处理函数")
+	}
+
+	token := client.client.Subscribe(topic, qos, func(mqttClient mqtt.Client, message mqtt.Message) {
+		handlerFunc(client, message.Topic(), message.Payload())
+	})
+
+	token.Wait()
+	return token.Error()
+}
+
+func (client *Client) Unsubscribe(topics ...string) error {
+	token := client.client.Unsubscribe(topics...)
+	token.Wait()
+	return token.Error()
+}

utils.go

@@ -0,0 +1,11 @@
+package mqtt_client
+
+import "strings"
+
+func isStringEmpty(s string) bool {
+	return strings.Trim(s, " ") == ""
+}
+
+func isStringNotEmpty(s string) bool {
+	return strings.Trim(s, " ") != ""
+}