remove subscriptions when a client disconnect

* modify

* remove

* modify

* modify

* remove no use

* add online/offline notification

* modify

* format log

* add reference

broker/broker.go

@@ -617,7 +617,9 @@ func (b *Broker) removeClient(c *client) {
 func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 	var subs []interface{}
 	var qoss []byte
+	b.mu.Lock()
 	err := b.topicsMgr.Subscribers([]byte(packet.TopicName), packet.Qos, &subs, &qoss)
+	b.mu.Unlock()
 	if err != nil {
 		log.Error("search sub client error,  ", zap.Error(err))
 		return

broker/client.go

@@ -51,11 +51,6 @@ type client struct {
 	rmsgs      []*packets.PublishPacket
 }
 
-type subInfo struct {
-	sub *subscription
-	num int
-}
-
 type subscription struct {
 	client *client
 	topic  string
@@ -117,6 +112,14 @@ func (c *client) readLoop() {
 				log.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 				msg := &Message{client: c, packet: DisconnectdPacket}
 				b.SubmitWork(msg)
+
+				// remove subscriptions related to that client
+				for topic, sub := range c.subMap {
+					t := []byte(topic)
+					c.topicsMgr.Unsubscribe(t, sub)
+					c.session.RemoveTopic(topic)
+					delete(c.subMap, topic)
+				}
 				return
 			}
 
@@ -218,7 +221,9 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 		}
 	}
 
+	c.mu.Lock()
 	err := c.topicsMgr.Subscribers([]byte(packet.TopicName), packet.Qos, &c.subs, &c.qoss)
+	c.mu.Unlock()
 	if err != nil {
 		log.Error("Error retrieving subscribers list: ", zap.String("ClientID", c.info.clientID))
 		return