feat: update

Bump golang.org/x/crypto from 0.14.0 to 0.17.0

.gitignore

@@ -7,3 +7,7 @@ log/*
 .vscode/settings.json
 .pre-commit-config.yaml
 hmq.exe
+*.sw*
+*.swo
+*.swp
+*.swn

broker/broker.go

@@ -2,6 +2,7 @@ package broker
 
 import (
 	"crypto/tls"
+	encJson "encoding/json"
 	"errors"
 	"fmt"
 	"net"
@@ -16,6 +17,8 @@ import (
 	"github.com/fhmq/hmq/pool"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
+
+	v5packets "github.com/eclipse/paho.golang/packets"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/net/websocket"
@@ -204,8 +207,8 @@ func (b *Broker) StartWebsocketListening() {
 func (b *Broker) wsHandler(ws *websocket.Conn) {
 	// io.Copy(ws, ws)
 	ws.PayloadType = websocket.BinaryFrame
-	err:=b.handleConnection(CLIENT, ws)
-	if err!=nil{
+	err := b.handleConnection(CLIENT, ws)
+	if err != nil {
 		ws.Close()
 	}
 }
@@ -258,9 +261,9 @@ func (b *Broker) StartClientListening(Tls bool) {
 		}
 
 		tmpDelay = ACCEPT_MIN_SLEEP
-		go func(){
-			err :=b.handleConnection(CLIENT, conn)
-			if err!=nil{
+		go func() {
+			err := b.handleConnection(CLIENT, conn)
+			if err != nil {
 				conn.Close()
 			}
 		}()
@@ -300,9 +303,9 @@ func (b *Broker) StartClusterListening() {
 		}
 		tmpDelay = ACCEPT_MIN_SLEEP
 
-		go func(){
-			err :=b.handleConnection(ROUTER, conn)
-			if err!=nil{
+		go func() {
+			err := b.handleConnection(ROUTER, conn)
+			if err != nil {
 				conn.Close()
 			}
 		}()
@@ -321,11 +324,128 @@ func (b *Broker) DisConnClientByClientId(clientId string) {
 	conn.Close()
 }
 
-func (b *Broker) handleConnection(typ int, conn net.Conn) error{
+func (b *Broker) handleV5Connection(typ int, conn net.Conn) error {
+	//process connect packet
+	cp, err := v5packets.ReadPacket(conn)
+	if err != nil {
+		return errors.New(fmt.Sprintf("read connect packet error:%v", err))
+	}
+	if cp == nil {
+		return errors.New("received nil packet")
+	}
+	if cp.Type != v5packets.CONNECT {
+		return errors.New("received msg that was not Connect")
+	}
+
+	msg := cp.Content.(*v5packets.Connect)
+
+	log.Info("read connect from ", getAdditionalLogFields(msg.ClientID, conn)...)
+
+	connack := &v5packets.Connack{
+		SessionPresent: msg.CleanStart,
+	}
+
+	if typ == CLIENT && !b.CheckConnectAuth(msg.ClientID, msg.Username, string(msg.Password)) {
+		connack.ReasonCode = v5packets.ConnackBadUsernameOrPassword
+		if _, err := connack.WriteTo(conn); err != nil {
+			return fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientID, conn)
+		}
+		return fmt.Errorf("connect packet CheckConnectAuth failed with connack.ReturnCode%v", connack.ReasonCode)
+	}
+
+	if _, err := connack.WriteTo(conn); err != nil {
+		return fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientID, conn)
+	}
+
+	willmsg := &v5packets.Publish{}
+	if msg.WillFlag {
+		willmsg.QoS = msg.WillQOS
+		willmsg.Topic = msg.WillTopic
+		willmsg.Retain = msg.WillRetain
+		willmsg.Payload = msg.WillMessage
+	} else {
+		willmsg = nil
+	}
+	info := info{
+		clientID:  msg.ClientID,
+		username:  msg.Username,
+		password:  msg.Password,
+		keepalive: msg.KeepAlive,
+		willMsg:   willmsg,
+	}
+
+	c := &client{
+		typ:    typ,
+		broker: b,
+		conn:   conn,
+		info:   info,
+	}
+
+	c.init()
+
+	if err := b.getSession(c, msg, connack); err != nil {
+		return fmt.Errorf("get session error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
+	}
+
+	cid := c.info.clientID
+
+	var exists bool
+	var old interface{}
+
+	switch typ {
+	case CLIENT:
+		old, exists = b.clients.Load(cid)
+		if exists {
+			if ol, ok := old.(*client); ok {
+				log.Warn("client exists, close old client", getAdditionalLogFields(ol.info.clientID, ol.conn)...)
+				ol.Close()
+			}
+		}
+		b.clients.Store(cid, c)
+
+		var pubPack = PubPacket{}
+		if willmsg != nil {
+			pubPack.TopicName = info.willMsg.TopicName
+			pubPack.Payload = info.willMsg.Payload
+		}
+
+		pubInfo := Info{
+			ClientID:  info.clientID,
+			Username:  info.username,
+			Password:  info.password,
+			Keepalive: info.keepalive,
+			WillMsg:   pubPack,
+		}
+
+		b.OnlineOfflineNotification(pubInfo, true, c.lastMsgTime)
+		{
+			b.Publish(&bridge.Elements{
+				ClientID:  msg.ClientIdentifier,
+				Username:  msg.Username,
+				Action:    bridge.Connect,
+				Timestamp: time.Now().Unix(),
+			})
+		}
+	case ROUTER:
+		old, exists = b.routes.Load(cid)
+		if exists {
+			if ol, ok := old.(*client); ok {
+				log.Warn("router exists, close old router", getAdditionalLogFields(ol.info.clientID, ol.conn)...)
+				ol.Close()
+			}
+		}
+		b.routes.Store(cid, c)
+	}
+
+	c.readLoop()
+	return nil
+}
+
+func (b *Broker) handleConnection(typ int, conn net.Conn) error {
 	//process connect packet
 	packet, err := packets.ReadPacket(conn)
 	if err != nil {
-		return errors.New(fmt.Sprintln("read connect packet error:%v",err))
+		return errors.New(fmt.Sprintf("read connect packet error:%v", err))
 	}
 	if packet == nil {
 		return errors.New("received nil packet")
@@ -343,21 +463,21 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) error{
 
 	if connack.ReturnCode != packets.Accepted {
 		if err := connack.Write(conn); err != nil {
-			return errors.New(fmt.Sprintln("send connack error:%v,clientID:%v,conn:%v",err,msg.ClientIdentifier,conn))
+			return fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
 		}
-		return errors.New(fmt.Sprintln("connect packet validate failed with connack.ReturnCode%v",connack.ReturnCode))
+		return fmt.Errorf("connect packet validate failed with connack.ReturnCode%v", connack.ReturnCode)
 	}
 
 	if typ == CLIENT && !b.CheckConnectAuth(msg.ClientIdentifier, msg.Username, string(msg.Password)) {
 		connack.ReturnCode = packets.ErrRefusedNotAuthorised
 		if err := connack.Write(conn); err != nil {
-			return errors.New(fmt.Sprintln("send connack error:%v,clientID:%v,conn:%v",err,msg.ClientIdentifier,conn))
+			return fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
 		}
-		return errors.New(fmt.Sprintln("connect packet CheckConnectAuth failed with connack.ReturnCode%v",connack.ReturnCode))
+		return fmt.Errorf("connect packet CheckConnectAuth failed with connack.ReturnCode%v", connack.ReturnCode)
 	}
 
 	if err := connack.Write(conn); err != nil {
-		return errors.New(fmt.Sprintln("send connack error:%v,clientID:%v,conn:%v",err,msg.ClientIdentifier,conn))
+		return fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
 	}
 
 	willmsg := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
@@ -388,7 +508,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) error{
 	c.init()
 
 	if err := b.getSession(c, msg, connack); err != nil {
-		return errors.New(fmt.Sprintln("get session error:%v,clientID:%v,conn:%v",err,msg.ClientIdentifier,conn))
+		return fmt.Errorf("get session error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
 	}
 
 	cid := c.info.clientID
@@ -407,7 +527,21 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) error{
 		}
 		b.clients.Store(cid, c)
 
-		b.OnlineOfflineNotification(cid, true)
+		var pubPack = PubPacket{}
+		if willmsg != nil {
+			pubPack.TopicName = info.willMsg.TopicName
+			pubPack.Payload = info.willMsg.Payload
+		}
+
+		pubInfo := Info{
+			ClientID:  info.clientID,
+			Username:  info.username,
+			Password:  info.password,
+			Keepalive: info.keepalive,
+			WillMsg:   pubPack,
+		}
+
+		b.OnlineOfflineNotification(pubInfo, true, c.lastMsgTime)
 		{
 			b.Publish(&bridge.Elements{
 				ClientID:  msg.ClientIdentifier,
@@ -441,7 +575,7 @@ func (b *Broker) ConnectToDiscovery() {
 			log.Error("Error trying to connect to route", zap.Error(err))
 			log.Debug("Connect to route timeout, retry...")
 
-			if 0 == tempDelay {
+			if tempDelay == 0 {
 				tempDelay = 1 * time.Second
 			} else {
 				tempDelay *= 2
@@ -513,7 +647,7 @@ func (b *Broker) connectRouter(id, addr string) {
 
 			log.Debug("Connect to route timeout, retry...")
 
-			if 0 == timeDelay {
+			if timeDelay == 0 {
 				timeDelay = 1 * time.Second
 			} else {
 				timeDelay *= 2
@@ -698,11 +832,33 @@ func (b *Broker) BroadcastUnSubscribe(topicsToUnSubscribeFrom []string) {
 	b.BroadcastSubOrUnsubMessage(unsub)
 }
 
-func (b *Broker) OnlineOfflineNotification(clientID string, online bool) {
+type OnlineOfflineMsg struct {
+	ClientID    string `json:"clientID"`
+	Online      bool   `json:"online"`
+	Timestamp   string `json:"timestamp"`
+	ClientInfo  Info   `json:"info"`
+	LastMsgTime int64  `json:"lastMsg"`
+}
+
+func (b *Broker) OnlineOfflineNotification(info Info, online bool, lastMsg int64) {
 	packet := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
-	packet.TopicName = "$SYS/broker/connection/clients/" + clientID
+	packet.TopicName = "$SYS/broker/connection/clients/" + info.ClientID
 	packet.Qos = 0
-	packet.Payload = []byte(fmt.Sprintf(`{"clientID":"%s","online":%v,"timestamp":"%s"}`, clientID, online, time.Now().UTC().Format(time.RFC3339)))
+
+	msg := OnlineOfflineMsg{
+		ClientID:    info.ClientID,
+		Online:      online,
+		Timestamp:   time.Now().UTC().Format(time.RFC3339),
+		ClientInfo:  info,
+		LastMsgTime: lastMsg,
+	}
+
+	if b, err := encJson.Marshal(msg); err != nil {
+		//This is a TERRIBLE situation, falling back to legacy format to not break API Contract
+		packet.Payload = []byte(fmt.Sprintf(`{"clientID":"%s","online":%v,"timestamp":"%s"}`, info.ClientID, online, time.Now().UTC().Format(time.RFC3339)))
+	} else {
+		packet.Payload = b
+	}
 
 	b.PublishMessage(packet)
 }

broker/client.go

@@ -79,6 +79,7 @@ type client struct {
 	mqueue         *queue.Queue
 	retryTimer     *time.Timer
 	retryTimerLock sync.Mutex
+	lastMsgTime    int64
 }
 
 type InflightStatus uint8
@@ -111,6 +112,19 @@ type info struct {
 	remoteIP  string
 }
 
+type PubPacket struct {
+	TopicName string `json:"topicName"`
+	Payload   []byte `json:"payload"`
+}
+
+type Info struct {
+	ClientID  string    `json:"clientId"`
+	Username  string    `json:"username"`
+	Password  []byte    `json:"password"`
+	Keepalive uint16    `json:"keepalive"`
+	WillMsg   PubPacket `json:"willMsg"`
+}
+
 type route struct {
 	remoteID  string
 	remoteUrl string
@@ -122,6 +136,7 @@ var (
 )
 
 func (c *client) init() {
+	c.lastMsgTime = time.Now().Unix() //mark the connection packet time as last time messaged
 	c.status = Connected
 	c.info.localIP, _, _ = net.SplitHostPort(c.conn.LocalAddr().String())
 	remoteAddr := c.conn.RemoteAddr()
@@ -185,6 +200,8 @@ func (c *client) readLoop() {
 			if _, isDisconnect := packet.(*packets.DisconnectPacket); isDisconnect {
 				c.info.willMsg = nil
 				c.cancelFunc()
+			} else {
+				c.lastMsgTime = time.Now().Unix()
 			}
 
 			msg := &Message{
@@ -842,8 +859,22 @@ func (c *client) Close() {
 
 	if c.typ == CLIENT {
 		b.BroadcastUnSubscribe(unSubTopics)
+
+		var pubPack = PubPacket{}
+		if c.info.willMsg != nil {
+			pubPack.TopicName = c.info.willMsg.TopicName
+			pubPack.Payload = c.info.willMsg.Payload
+		}
+
+		pubInfo := Info{
+			ClientID:  c.info.clientID,
+			Username:  c.info.username,
+			Password:  c.info.password,
+			Keepalive: c.info.keepalive,
+			WillMsg:   pubPack,
+		}
 		//offline notification
-		b.OnlineOfflineNotification(c.info.clientID, false)
+		b.OnlineOfflineNotification(pubInfo, false, c.lastMsgTime)
 	}
 
 	if c.info.willMsg != nil {

broker/config.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"os"
 
 	"github.com/fhmq/hmq/logger"
@@ -22,6 +21,7 @@ type Config struct {
 	Worker   int       `json:"workerNum"`
 	HTTPPort string    `json:"httpPort"`
 	Host     string    `json:"host"`
+	V5Port   string    `json:"v5Port"`
 	Port     string    `json:"port"`
 	Cluster  RouteInfo `json:"cluster"`
 	Router   string    `json:"router"`
@@ -89,6 +89,8 @@ func ConfigureConfig(args []string) (*Config, error) {
 	fs.StringVar(&config.HTTPPort, "hp", "8080", "Port to listen on.")
 	fs.StringVar(&config.Port, "port", "1883", "Port to listen on.")
 	fs.StringVar(&config.Port, "p", "1883", "Port to listen on.")
+	fs.StringVar(&config.V5Port, "v5port", "", "Port v5 to listen on.")
+	fs.StringVar(&config.V5Port, "p5", "", "Port v5 to listen on.")
 	fs.StringVar(&config.Host, "host", "0.0.0.0", "Network host to listen on")
 	fs.StringVar(&config.Cluster.Port, "cp", "", "Cluster port from which members can connect.")
 	fs.StringVar(&config.Cluster.Port, "clusterport", "", "Cluster port from which members can connect.")
@@ -144,7 +146,7 @@ func ConfigureConfig(args []string) (*Config, error) {
 
 func LoadConfig(filename string) (*Config, error) {
 
-	content, err := ioutil.ReadFile(filename)
+	content, err := os.ReadFile(filename)
 	if err != nil {
 		// log.Error("Read config file error: ", zap.Error(err))
 		return nil, err
@@ -231,7 +233,7 @@ func NewTLSConfig(tlsInfo TLSInfo) (*tls.Config, error) {
 	}
 	// Add in CAs if applicable.
 	if tlsInfo.CaFile != "" {
-		rootPEM, err := ioutil.ReadFile(tlsInfo.CaFile)
+		rootPEM, err := os.ReadFile(tlsInfo.CaFile)
 		if err != nil || rootPEM == nil {
 			return nil, err
 		}

broker/http.go

@@ -4,10 +4,24 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
+const (
+	CONNECTIONS	= "api/v1/connections"
+)
+
+type ConnClient struct {
+	Info				`json:"info"`
+	LastMsgTime	int64		`json:"lastMsg"`
+}
+
+type resp struct {
+	Code	int		`json:"code,omitempty"`
+	Clients	[]ConnClient	`json:"clients,omitempty"`
+}
+
 func InitHTTPMoniter(b *Broker) {
 	gin.SetMode(gin.ReleaseMode)
 	router := gin.Default()
-	router.DELETE("api/v1/connections/:clientid", func(c *gin.Context) {
+	router.DELETE(CONNECTIONS + "/:clientid", func(c *gin.Context) {
 		clientid := c.Param("clientid")
 		cli, ok := b.clients.Load(clientid)
 		if ok {
@@ -16,10 +30,35 @@ func InitHTTPMoniter(b *Broker) {
 				conn.Close()
 			}
 		}
-		resp := map[string]int{
-			"code": 0,
-		}
-		c.JSON(200, &resp)
+		r := resp{Code: 0}
+		c.JSON(200, &r)
+	})
+	router.GET(CONNECTIONS, func(c *gin.Context) {
+		conns := make([]ConnClient, 0)
+		b.clients.Range(func (k, v interface{}) bool {
+			cl, _ := v.(*client)
+			var pubPack = PubPacket{}
+			if cl.info.willMsg != nil {
+				pubPack.TopicName = cl.info.willMsg.TopicName
+				pubPack.Payload = cl.info.willMsg.Payload
+			}
+
+			msg := ConnClient{
+				Info: Info{
+					ClientID: cl.info.clientID,
+					Username: cl.info.username,
+					Password: cl.info.password,
+					Keepalive: cl.info.keepalive,
+					WillMsg: pubPack,
+				},
+				LastMsgTime: cl.lastMsgTime,
+			}
+
+			conns = append(conns, msg)
+			return true
+		})
+		r := resp{Clients: conns}
+		c.JSON(200, &r)
 	})
 
 	router.Run(":" + b.config.HTTPPort)

conf/hmq.config

@@ -1,6 +1,7 @@
 {
 	"workerNum": 4096,
 	"port": "1883",
+	"v5Port": "1885",
 	"host": "0.0.0.0",
 	"cluster": {
 		"host": "0.0.0.0",
@@ -11,6 +12,7 @@
 	"tlsHost": "0.0.0.0",
 	"wsPort": "1888",
 	"wsPath": "/ws",
+	"ws5Path": "/ws5",
 	"wsTLS": false,
 	"tlsInfo": {
 		"verify": true,

go.mod

@@ -1,18 +1,19 @@
 module github.com/fhmq/hmq
 
-go 1.18
+go 1.21
 
 require (
 	github.com/Shopify/sarama v1.38.1
 	github.com/bitly/go-simplejson v0.5.0
 	github.com/cespare/xxhash/v2 v2.1.2
 	github.com/eapache/queue v1.1.0
-	github.com/eclipse/paho.mqtt.golang v1.4.2
+	github.com/eclipse/paho.golang v0.12.0
+	github.com/eclipse/paho.mqtt.golang v1.4.3
 	github.com/gin-gonic/gin v1.9.1
 	github.com/google/uuid v1.3.0
 	github.com/json-iterator/go v1.1.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/stretchr/testify v1.8.3
+	github.com/stretchr/testify v1.8.4
 	go.uber.org/zap v1.24.0
 	golang.org/x/net v0.17.0
 )
@@ -31,7 +32,7 @@ require (
 	github.com/go-playground/validator/v10 v10.14.0 // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
-	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
 	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
@@ -39,9 +40,8 @@ require (
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.3 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
-	github.com/klauspost/compress v1.15.15 // indirect
+	github.com/klauspost/compress v1.15.14 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
-	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.2.4 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -50,11 +50,11 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
-	github.com/rogpeppe/go-internal v1.10.0 // indirect
+	github.com/rogpeppe/go-internal v1.12.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.11 // indirect
-	go.uber.org/atomic v1.10.0 // indirect
-	go.uber.org/multierr v1.9.0 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/crypto v0.14.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect

go.sum

@@ -1,7 +1,9 @@
 github.com/Shopify/sarama v1.38.1 h1:lqqPUPQZ7zPqYlWpTh+LQ9bhYNu2xJL6k1SJN4WVe2A=
 github.com/Shopify/sarama v1.38.1/go.mod h1:iwv9a67Ha8VNa+TifujYoWGxWnu2kNVAQdSdZ4X2o5g=
 github.com/Shopify/toxiproxy/v2 v2.5.0 h1:i4LPT+qrSlKNtQf5QliVjdP08GyAH8+BUIc9gT0eahc=
+github.com/Shopify/toxiproxy/v2 v2.5.0/go.mod h1:yhM2epWtAmel9CB8r2+L+PCmhH6yH2pITaPAo7jxJl0=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
+github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkNEM+Y=
 github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
@@ -14,7 +16,6 @@ github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/chenzhuoyu/base64x v0.0.0-20211019084208-fb5309c8db06/go.mod h1:DH46F32mSOjUmXrMHnKwZdA8wcEefY7UVqBKYGjpdQY=
 github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 h1:qSGYFH7+jGhDF8vLC+iwCD4WpbV1EBDSzWkJODFLams=
 github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311/go.mod h1:b583jCggY9gE99b6G5LEC39OIiVsWj+R97kbl5odCEk=
-github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -24,9 +25,12 @@ github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6 h1:8yY/I9
 github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
-github.com/eclipse/paho.mqtt.golang v1.4.2 h1:66wOzfUHSSI1zamx7jR6yMEI5EuHnT1G6rNA5PM12m4=
-github.com/eclipse/paho.mqtt.golang v1.4.2/go.mod h1:JGt0RsEwEX+Xa/agj90YJ9d9DH2b7upDZMK9HRbFvCA=
+github.com/eclipse/paho.golang v0.12.0 h1:EXQFJbJklDnUqW6lyAknMWRhM2NgpHxwrrL8riUmp3Q=
+github.com/eclipse/paho.golang v0.12.0/go.mod h1:TSDCUivu9JnoR9Hl+H7sQMcHkejWH2/xKK1NJGtLbIE=
+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/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
+github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/gabriel-vasile/mimetype v1.4.2 h1:w5qFW6JKBz9Y393Y4q372O9A7cUSequkh1Q7OhCmWKU=
 github.com/gabriel-vasile/mimetype v1.4.2/go.mod h1:zApsH/mKG4w07erKIaJPFiX0Tsq9BFQgN3qGY5GnNgA=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
@@ -34,6 +38,7 @@ github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm
 github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg=
 github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU=
 github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
+github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
 github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
 github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
@@ -45,17 +50,16 @@ github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MG
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
-github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
-github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
-github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
@@ -75,12 +79,13 @@ github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZ
 github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
-github.com/klauspost/compress v1.15.15 h1:EF27CXIuDsYJ6mmvtBRlEuB2UVOqHG1tAXgZ7yIO+lw=
-github.com/klauspost/compress v1.15.15/go.mod h1:ZcK2JAFqKOpnBlxcLsJzYfrS9X1akm9fHZNnD9+Vo/4=
+github.com/klauspost/compress v1.15.14 h1:i7WCKDToww0wA+9qrUZ1xOjp218vfFo3nTU6UHp+gOc=
+github.com/klauspost/compress v1.15.14/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
 github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
 github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/leodido/go-urn v1.2.4 h1:XlAE/cm/ms7TE/VMVoduSpNBoyc2dOxHs5MZSwAN63Q=
@@ -99,12 +104,13 @@ github.com/pelletier/go-toml/v2 v2.0.8/go.mod h1:vuYfssBdrU2XDZ9bYydBu6t+6a6PYNc
 github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
 github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
-github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
-github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
+github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -115,17 +121,19 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/stretchr/testify v1.8.3 h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY=
 github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.11 h1:BMaWp1Bb6fHwEtbplGBGJ498wD+LKlNSl25MjdZY4dU=
 github.com/ugorji/go/codec v1.2.11/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
-go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
-go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
-go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
-go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=
-go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ=
+go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
+go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
+go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
+go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
 go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
@@ -136,15 +144,13 @@ golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0
 golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
 golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220725212005-46097bf591d3/go.mod h1:AaygXjzTFtRAg2ttMY5RMuhpJ3cNnI0XpyFJD1iQRSM=
 golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
 golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.4.0 h1:zxkM55ReGkDlKSM+Fu41A+zmbZuaPVbGMzvvdUPznYQ=
+golang.org/x/sync v0.4.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -161,13 +167,13 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
 google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=