packet

broker/broker.go

@@ -342,6 +342,7 @@ func (b *Broker) connectRouter(url, remoteID string) {
 		route: route,
 		info:  info,
 	}
+	c.init()
 	b.remotes.Store(cid, c)
 	c.SendConnect()
 	c.SendInfo()

broker/client.go

@@ -33,6 +33,8 @@ type client struct {
 	conn   net.Conn
 	info   info
 	route  *route
+	status int
+	smu    sync.RWMutex
 	subs   map[string]*subscription
 	rsubs  map[string]*subInfo
 }
@@ -69,6 +71,9 @@ var (
 )
 
 func (c *client) init() {
+	c.smu.Lock()
+	defer c.smu.Unlock()
+	c.status = Connected
 	typ := c.typ
 	if typ == ROUTER {
 		c.rsubs = make(map[string]*subInfo)
@@ -162,8 +167,11 @@ func ProcessMessage(msg *Message) {
 }
 
 func (c *client) ProcessPublish(packet *packets.PublishPacket) {
-	topic := packet.TopicName
+	if c.status == Disconnected {
+		return
+	}
 
+	topic := packet.TopicName
 	if !c.CheckTopicAuth(PUB, topic) {
 		log.Error("Pub Topics Auth failed, ", topic)
 		return
@@ -198,6 +206,9 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) {
 }
 
 func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
+	if c.status == Disconnected {
+		return
+	}
 
 	b := c.broker
 	if b == nil {
@@ -282,6 +293,10 @@ func getQueueSubscribeNum(qsubs []*subscription) int {
 }
 
 func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
+	if c.status == Disconnected {
+		return
+	}
+
 	b := c.broker
 	if b == nil {
 		return
@@ -375,6 +390,9 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 }
 
 func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
+	if c.status == Disconnected {
+		return
+	}
 	b := c.broker
 	if b == nil {
 		return
@@ -437,6 +455,9 @@ func (c *client) unsubscribe(sub *subscription) {
 }
 
 func (c *client) ProcessPing() {
+	if c.status == Disconnected {
+		return
+	}
 	resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
 	err := c.WriterPacket(resp)
 	if err != nil {
@@ -446,6 +467,13 @@ func (c *client) ProcessPing() {
 }
 
 func (c *client) Close() {
+	c.smu.Lock()
+	c.status = Disconnected
+	if c.conn != nil {
+		c.conn.Close()
+		c.conn = nil
+	}
+	c.smu.Unlock()
 	b := c.broker
 	subs := c.subs
 	if b != nil {
@@ -463,10 +491,6 @@ func (c *client) Close() {
 			b.PublishMessage(c.info.willMsg)
 		}
 	}
-	if c.conn != nil {
-		c.conn.Close()
-		c.conn = nil
-	}
 }
 
 func (c *client) WriterPacket(packet packets.ControlPacket) error {

broker/info.go

@@ -11,6 +11,9 @@ import (
 )
 
 func (c *client) SendInfo() {
+	if c.status == Disconnected {
+		return
+	}
 	url := c.info.localIP + ":" + c.broker.config.Cluster.Port
 
 	infoMsg := NewInfo(c.broker.id, url, false)
@@ -37,7 +40,9 @@ func (c *client) StartPing() {
 }
 
 func (c *client) SendConnect() {
-
+	if c.status == Disconnected {
+		return
+	}
 	m := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
 
 	m.CleanSession = true