modify close old connect connection logic

broker/broker.go

@@ -292,22 +292,32 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	switch typ {
 	case CLIENT:
 		msgPool = MSGPool[idx%MessagePoolNum].GetPool()
+		c.mp = msgPool
 		old, exist = b.clients.Load(cid)
+		if exist {
+			log.Warn("client exist, close old...")
+			ol, ok := old.(*client)
+			if ok {
+				ol.Close()
+			}
+		}
 		b.clients.Store(cid, c)
 	case ROUTER:
 		msgPool = MSGPool[(MessagePoolNum + idx)].GetPool()
+		c.mp = msgPool
 		old, exist = b.routes.Load(cid)
+		if exist {
+			log.Warn("router exist, close old...")
+			ol, ok := old.(*client)
+			if ok {
+				msg := &Message{client: c, packet: DisconnectdPacket}
+				ol.mp.queue <- msg
+			}
+		}
 		b.routes.Store(cid, c)
 	}
 
-	if exist {
+	c.readLoop()
-		log.Warn("client or routers exist, close old...")
-		ol, ok := old.(*client)
-		if ok {
-			ol.Close()
-		}
-	}
-	c.readLoop(msgPool)
 }
 
 func (b *Broker) ConnectToRouters() {

broker/client.go

@@ -35,6 +35,7 @@ type client struct {
 	route  *route
 	status int
 	smu    sync.RWMutex
+	mp     *MessagePool
 	subs   map[string]*subscription
 	rsubs  map[string]*subInfo
 }
@@ -84,8 +85,9 @@ func (c *client) init() {
 	c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0]
 }
 
-func (c *client) readLoop(msgPool *MessagePool) {
+func (c *client) readLoop() {
 	nc := c.conn
+	msgPool := c.mp
 	if nc == nil || msgPool == nil {
 		return
 	}