collab/hmq
3
0
Fork 0
mirror of https://github.com/fhmq/hmq.git synced 2026-08-29 14:03:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Optimize broker runtime and update dependencies

Browse Source
This commit is contained in:
zhouyy
2026-05-31 09:53:43 +08:00
parent c609511826
commit bd2e23ed9f
12 changed files with 189 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/broker.go
+7 -7
View File
@@ -395,7 +395,7 @@ 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.Sprintf("read connect packet error:%v", err))
return fmt.Errorf("read connect packet error: %w", err)
}
if packet == nil {
return errors.New("received nil packet")
@@ -413,21 +413,21 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) error {
if connack.ReturnCode != packets.Accepted {
if err := connack.Write(conn); err != nil {
return fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
return fmt.Errorf("send connack error: %w, clientID: %v, conn: %v", err, msg.ClientIdentifier, conn)
}
return fmt.Errorf("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 fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
return fmt.Errorf("send connack error: %w, clientID: %v, conn: %v", err, msg.ClientIdentifier, conn)
}
return fmt.Errorf("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 fmt.Errorf("send connack error:%v,clientID:%v,conn:%v", err, msg.ClientIdentifier, conn)
return fmt.Errorf("send connack error: %w, clientID: %v, conn: %v", err, msg.ClientIdentifier, conn)
}
willmsg := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
@@ -458,7 +458,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) error {
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)
return fmt.Errorf("get session error: %w, clientID: %v, conn: %v", err, msg.ClientIdentifier, conn)
}
cid := c.info.clientID
broker/client.go
+11 -11
View File
@@ -1,10 +1,8 @@
package broker
import (
"bytes"
"context"
"errors"
"math/rand"
"net"
"reflect"
"regexp"
@@ -132,7 +130,6 @@ type route struct {
var (
DisconnectedPacket = packets.NewControlPacket(packets.Disconnect).(*packets.DisconnectPacket)
r = rand.New(rand.NewSource(time.Now().UnixNano()))
)
func (c *client) init() {
@@ -229,11 +226,12 @@ func extractPacketFields(msgPacket packets.ControlPacket) []string {
break
case *packets.SubscribePacket:
packet := msgPacket.(*packets.SubscribePacket)
fields = append(fields, packet.Topics...)
case *packets.SubackPacket:
case *packets.UnsubscribePacket:
packet := msgPacket.(*packets.UnsubscribePacket)
fields = append(fields, packet.Topics...)
break
}
return fields
@@ -259,7 +257,7 @@ func validatePacketFields(msgPacket packets.ControlPacket) (validFields bool) {
// If a receiver (Server or Client) receives a Control Packet containing U+0000
// it MUST close the Network Connection
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.pdf page 14
if bytes.ContainsAny([]byte(field), "\u0000") {
if strings.ContainsRune(field, '\u0000') {
validFields = false
return
}
@@ -319,9 +317,8 @@ func ProcessMessage(msg *Message) {
c.inflightMu.Unlock()
case *packets.PubrecPacket:
packet := ca.(*packets.PubrecPacket)
c.inflightMu.RLock()
c.inflightMu.Lock()
ielem, found := c.inflight[packet.MessageID]
c.inflightMu.RUnlock()
if found {
if ielem.status == Publish {
ielem.status = Pubrel
@@ -332,6 +329,7 @@ func ProcessMessage(msg *Message) {
} else {
log.Error("The PUBREC PacketId is not found.", zap.Uint16("MessageID", packet.MessageID))
}
c.inflightMu.Unlock()
pubrel := packets.NewControlPacket(packets.Pubrel).(*packets.PubrelPacket)
pubrel.MessageID = packet.MessageID
@@ -519,7 +517,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
}
if len(qsub) > 0 {
idx := r.Intn(len(qsub))
idx := int(time.Now().UnixNano() % int64(len(qsub)))
sub := c.subs[qsub[idx]].(*subscription)
publish(sub, packet)
}
@@ -552,7 +550,8 @@ func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
suback.MessageID = packet.MessageID
var retcodes []byte
retcodes := make([]byte, 0, len(subTopics))
retainedMsgs := c.rmsgs[:0]
for i, topic := range subTopics {
t := topic
@@ -612,8 +611,9 @@ func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
_ = c.session.AddTopic(t, qoss[i])
retcodes = append(retcodes, rqos)
_ = c.topicsMgr.Retained([]byte(topic), &c.rmsgs)
_ = c.topicsMgr.Retained([]byte(topic), &retainedMsgs)
}
c.rmsgs = retainedMsgs[:0]
suback.ReturnCodes = retcodes
@@ -626,7 +626,7 @@ func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
go b.BroadcastSubOrUnsubMessage(packet)
//process retain message
for _, rm := range c.rmsgs {
for _, rm := range retainedMsgs {
if err := c.WriterPacket(rm); err != nil {
log.Error("Error publishing retained message:", zap.Any("err", err), zap.String("ClientID", c.info.clientID))
} else {
broker/comm.go
+37 -14
View File
@@ -199,24 +199,38 @@ func (c *client) resetRetryTimer() {
func (c *client) retryDelivery() {
c.resetRetryTimer()
c.inflightMu.RLock()
ilen := len(c.inflight)
c.mu.Lock()
if c.conn == nil || ilen == 0 { //Reset timer when client offline OR inflight is empty
c.inflightMu.RUnlock()
c.mu.Unlock()
connLost := c.conn == nil
c.mu.Unlock()
if connLost {
return
}
c.mu.Unlock()
// copy the to be retried elements out of the map to only hold the lock for a short time and use the new slice later to iterate
// through them
toRetryEle := make([]*inflightElem, 0, ilen)
for _, infEle := range c.inflight {
toRetryEle = append(toRetryEle, infEle)
c.inflightMu.Lock()
ilen := len(c.inflight)
if ilen == 0 {
c.inflightMu.Unlock()
return
}
c.inflightMu.RUnlock()
type retryElem struct {
id uint16
status InflightStatus
packet *packets.PublishPacket
timestamp int64
}
toRetryEle := make([]retryElem, 0, ilen)
for id, infEle := range c.inflight {
toRetryEle = append(toRetryEle, retryElem{
id: id,
status: infEle.status,
packet: infEle.packet,
timestamp: infEle.timestamp,
})
}
c.inflightMu.Unlock()
now := time.Now().Unix()
for _, infEle := range toRetryEle {
@@ -224,12 +238,12 @@ func (c *client) retryDelivery() {
if age >= retryInterval {
if infEle.status == Publish {
c.WriterPacket(infEle.packet)
infEle.timestamp = now
c.updateInflightTimestamp(infEle.id, infEle.status, now)
} else if infEle.status == Pubrel {
pubrel := packets.NewControlPacket(packets.Pubrel).(*packets.PubrelPacket)
pubrel.MessageID = infEle.packet.MessageID
c.WriterPacket(pubrel)
infEle.timestamp = now
c.updateInflightTimestamp(infEle.id, infEle.status, now)
}
} else {
if age < 0 {
@@ -240,3 +254,12 @@ func (c *client) retryDelivery() {
}
c.ensureRetryTimer()
}
func (c *client) updateInflightTimestamp(id uint16, status InflightStatus, timestamp int64) {
c.inflightMu.Lock()
defer c.inflightMu.Unlock()
if infEle, ok := c.inflight[id]; ok && infEle.status == status {
infEle.timestamp = timestamp
}
}
broker/lib/sessions/memprovider.go
+4
View File
@@ -53,10 +53,14 @@ func (this *memProvider) Save(id string) error {
}
func (this *memProvider) Count() int {
this.mu.RLock()
defer this.mu.RUnlock()
return len(this.st)
}
func (this *memProvider) Close() error {
this.mu.Lock()
defer this.mu.Unlock()
this.st = make(map[string]*Session)
return nil
}