jl-iot/mqtt/config.go

package mqtt

import (
	"fmt"
	"log/slog"
	"net/url"
	"time"

	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

type IotMqttConfig struct {
	AppId             string // 所属应用编号
	BrokerUrl         string // Broker地址
	ClientId          string // 客户端ID
	Username          string // 用户名
	Password          string // 密码
	KeepAlive         uint16 // 保活时间间隔,单位s,默认为60
	ConnectRetryDelay uint16 // 连接重试延时,单位s,默认为3
	ConnectTimeout    uint16 // 连接操作超时,单位s,默认为3
}

func (c *IotMqttConfig) tryInto() (*autopaho.ClientConfig, error) {
	addr, err := url.Parse(c.BrokerUrl)
	if err != nil {
		return nil, fmt.Errorf("Mqtt.Address格式错误, %s: %w", c.BrokerUrl, err)
	}
	if c.KeepAlive == 0 {
		c.KeepAlive = 60
	}
	if c.ConnectRetryDelay == 0 {
		c.ConnectRetryDelay = 3
	}
	if c.ConnectTimeout == 0 {
		c.ConnectTimeout = 3
	}
	cc := &autopaho.ClientConfig{
		BrokerUrls: []*url.URL{
			addr,
		},
		KeepAlive:         c.KeepAlive,
		ConnectRetryDelay: time.Duration(c.ConnectRetryDelay) * time.Second,
		ConnectTimeout:    time.Duration(c.ConnectTimeout) * time.Second,
		OnConnectionUp: func(*autopaho.ConnectionManager, *paho.Connack) {
			slog.Info("MQTT连接成功", "url", c.BrokerUrl)
			subIotQc()
		},
		OnConnectError: func(err error) {
			slog.Error("MQTT连接失败", "url", c.BrokerUrl, "error", err)
		},
		ClientConfig: paho.ClientConfig{
			ClientID:      c.ClientId,
			Router:        paho.NewStandardRouter(),
			OnClientError: func(err error) { slog.Error("MQTT客户端发生错误", "clientId", c.ClientId, "err", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				slog.Error("MQTT连接断开", "clientId", c.ClientId, "reasonCode", d.ReasonCode, "properties", d.Properties)
			},
		},
	}
	cc.SetUsernamePassword(c.Username, []byte(c.Password))
	return cc, nil
}