mqp

broker/broker.go

@@ -6,6 +6,7 @@ import (
 	"hmq/packets"
 	"net"
 	"net/http"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -20,9 +21,9 @@ type Broker struct {
 	config    *Config
 	tlsConfig *tls.Config
 	AclConfig *acl.ACLConfig
-	clients   cMap
+	clients   sync.Map
-	routes    cMap
+	routes    sync.Map
-	remotes   cMap
+	remotes   sync.Map
 	sl        *Sublist
 	rl        *RetainList
 	queues    map[string]int
@@ -35,9 +36,6 @@ func NewBroker(config *Config) (*Broker, error) {
 		sl:     NewSublist(),
 		rl:     NewRetainList(),
 		queues: make(map[string]int),
-		clients: NewClientMap(),
-		routes:  NewClientMap(),
-		remotes: NewClientMap(),
 	}
 	if b.config.TlsPort != "" {
 		tlsconfig, err := NewTLSConfig(b.config.TlsInfo)
@@ -217,20 +215,31 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	}
 	c.init()
 
+	cid := c.info.clientID
+
 	var msgPool *MessagePool
-	var exist bool
+	var loaded bool
 	var old *client
-	cid := string(c.info.clientID)
+
-	if typ == CLIENT {
+	switch typ {
-		old, exist = b.clients.Update(cid, c)
+	case CLIENT:
 		msgPool = MSGPool[idx%MessagePoolNum].GetPool()
-	} else if typ == ROUTER {
+		old, loaded = b.clients.LoadOrStore(cid)
-		old, exist = b.routes.Update(cid, c)
+		if !loaded {
-		msgPool = MSGPool[MessagePoolNum].GetPool()
+			b.clients.Store(cid, c)
+			old.(*client).Close()
 		}
-	if exist {
+	case ROUTER:
+		msgPool = MSGPool[MessagePoolNum].GetPool()
+		old, loaded = b.routes.LoadOrStore(cid)
+		if !loaded {
+			b.routes.Store(cid, c)
+		}
+	}
+
+	if !loaded {
 		log.Warn("client or routers exists, close old...")
-		old.Close()
+		old.(*client).Close()
 	}
 	c.readLoop(msgPool)
 }
@@ -267,7 +276,7 @@ func (b *Broker) connectRouter(url, remoteID string) {
 			route: route,
 			info:  info,
 		}
-		b.remotes.Set(cid, c)
+		b.remotes.Store(cid, c)
 		c.SendConnect()
 		c.SendInfo()
 		// s.createRemote(conn, route)
@@ -278,29 +287,35 @@ func (b *Broker) connectRouter(url, remoteID string) {
 
 func (b *Broker) CheckRemoteExist(remoteID, url string) bool {
 	exist := false
-	remotes := b.remotes.Items()
+	remotes := b.remotes.Range(func(key, value interface{}) bool {
-	for _, v := range remotes {
+		v, ok := value.(*client)
+		if ok {
 			if v.route.remoteUrl == url {
 				// if v.route.remoteID == "" || v.route.remoteID != remoteID {
 				v.route.remoteID = remoteID
 				// }
 				exist = true
-			break
+				return false
 			}
 		}
+		return true
+	})
 	return exist
 }
 
 func (b *Broker) SendLocalSubsToRouter(c *client) {
-	clients := b.clients.Items()
 	subInfo := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
-	for _, client := range clients {
+	b.clients.Range(func(key, value interface{}) bool {
+		client, ok := value.(*clients)
+		if ok {
 			subs := client.subs
 			for _, sub := range subs {
 				subInfo.Topics = append(subInfo.Topics, string(sub.topic))
 				subInfo.Qoss = append(subInfo.Qoss, sub.qos)
 			}
 		}
+		return true
+	})
 	err := c.WriterPacket(subInfo)
 	if err != nil {
 		log.Error("Send localsubs To Router error :", err)
@@ -308,21 +323,28 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 }
 
 func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacket) {
-	remotes := b.remotes.Items()
+	b.remotes.Range(func(key, value interface{}) bool {
-	for _, r := range remotes {
+		r, ok := value.(*client)
+		if ok {
 			if r.route.remoteID == remoteID {
-			continue
+				return true
 			}
 			r.WriterPacket(msg)
 		}
+		return true
+
+	})
 	// log.Info("BroadcastInfoMessage success ")
 }
 
 func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
-	remotes := b.remotes.Items()
+	b.remotes.Range(func(key, value interface{}) bool {
-	for _, r := range remotes {
+		r, ok := value.(*client)
-		r.WriterPacket(packet)
+		if ok {
+			r.WriterPacket(msg)
 		}
+		return true
+	})
 	// log.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
 }