modify readme

README.md

@@ -28,7 +28,7 @@ Broker Options:
 
 Logging Options:
     -d, --debug <bool>                Enable debugging output (default false)
-    -D                                Debug enabled
+    -D                                Debug and trace
 
 Cluster Options:
     -r,  --router  <rurl>             Router who maintenance cluster info

broker/auth.go

@@ -3,10 +3,13 @@
 package broker
 
 import (
-	"github.com/fhmq/hmq/lib/acl"
-	"github.com/fsnotify/fsnotify"
-	"go.uber.org/zap"
 	"strings"
+
+	"github.com/fhmq/hmq/lib/acl"
+
+	"go.uber.org/zap"
+
+	"github.com/fsnotify/fsnotify"
 )
 
 const (
@@ -41,10 +44,10 @@ func (b *Broker) handleFsEvent(event fsnotify.Event) error {
 	case b.config.AclConf:
 		if event.Op&fsnotify.Write == fsnotify.Write ||
 			event.Op&fsnotify.Create == fsnotify.Create {
-			log.Info("text:handling acl config change event:", zap.String("filename", event.Name))
+			brokerLog.Info("text:handling acl config change event:", zap.String("filename", event.Name))
 			aclconfig, err := acl.AclConfigLoad(event.Name)
 			if err != nil {
-				log.Error("aclconfig change failed, load acl conf error: ", zap.Error(err))
+				brokerLog.Error("aclconfig change failed, load acl conf error: ", zap.Error(err))
 				return err
 			}
 			b.AclConfig = aclconfig
@@ -57,24 +60,24 @@ func (b *Broker) StartAclWatcher() {
 	go func() {
 		wch, e := fsnotify.NewWatcher()
 		if e != nil {
-			log.Error("start monitor acl config file error,", zap.Error(e))
+			brokerLog.Error("start monitor acl config file error,", zap.Error(e))
 			return
 		}
 		defer wch.Close()
 
 		for _, i := range watchList {
 			if err := wch.Add(i); err != nil {
-				log.Error("start monitor acl config file error,", zap.Error(err))
+				brokerLog.Error("start monitor acl config file error,", zap.Error(err))
 				return
 			}
 		}
-		log.Info("watching acl config file change...")
+		brokerLog.Info("watching acl config file change...")
 		for {
 			select {
 			case evt := <-wch.Events:
 				b.handleFsEvent(evt)
 			case err := <-wch.Errors:
-				log.Error("error:", zap.Error(err))
+				brokerLog.Error("error:", zap.Error(err))
 			}
 		}
 	}()

broker/broker.go

@@ -11,17 +11,18 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/eclipse/paho.mqtt.golang/packets"
 	"github.com/fhmq/hmq/lib/acl"
 	"github.com/fhmq/hmq/pool"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
 	"github.com/shirou/gopsutil/mem"
 	"go.uber.org/zap"
+
 	"golang.org/x/net/websocket"
 )
 
-const (
-	MessagePoolNum        = 1024
-	MessagePoolMessageNum = 1024
+var (
+	brokerLog *zap.Logger
 )
 
 type Message struct {
@@ -42,19 +43,10 @@ type Broker struct {
 	remotes     sync.Map
 	nodes       map[string]interface{}
 	clusterPool chan *Message
+	messagePool chan *Message
 	sl          *Sublist
 	rl          *RetainList
 	queues      map[string]int
-	// messagePool []chan *Message
-}
-
-func newMessagePool() []chan *Message {
-	pool := make([]chan *Message, 0)
-	for i := 0; i < MessagePoolNum; i++ {
-		ch := make(chan *Message, MessagePoolMessageNum)
-		pool = append(pool, ch)
-	}
-	return pool
 }
 
 func NewBroker(config *Config) (*Broker, error) {
@@ -67,12 +59,12 @@ func NewBroker(config *Config) (*Broker, error) {
 		nodes:       make(map[string]interface{}),
 		queues:      make(map[string]int),
 		clusterPool: make(chan *Message),
-		// messagePool: newMessagePool(),
+		messagePool: make(chan *Message),
 	}
 	if b.config.TlsPort != "" {
 		tlsconfig, err := NewTLSConfig(b.config.TlsInfo)
 		if err != nil {
-			log.Error("new tlsConfig error", zap.Error(err))
+			brokerLog.Error("new tlsConfig error", zap.Error(err))
 			return nil, err
 		}
 		b.tlsConfig = tlsconfig
@@ -80,7 +72,7 @@ func NewBroker(config *Config) (*Broker, error) {
 	if b.config.Acl {
 		aclconfig, err := acl.AclConfigLoad(b.config.AclConf)
 		if err != nil {
-			log.Error("Load acl conf error", zap.Error(err))
+			brokerLog.Error("Load acl conf error", zap.Error(err))
 			return nil, err
 		}
 		b.AclConfig = aclconfig
@@ -89,14 +81,13 @@ func NewBroker(config *Config) (*Broker, error) {
 	return b, nil
 }
 
-func (b *Broker) SubmitWork(msg *Message) {
-	if b.wpool == nil {
-		b.wpool = pool.New(b.config.Worker)
-	}
-
-	if msg.client.typ == CLUSTER {
-		b.clusterPool <- msg
-	} else {
+func (b *Broker) StartDispatcher() {
+	for {
+		msg, ok := <-b.messagePool
+		if !ok {
+			brokerLog.Error("read message from client channel error")
+			return
+		}
 		b.wpool.Submit(func() {
 			ProcessMessage(msg)
 		})
@@ -106,9 +97,10 @@ func (b *Broker) SubmitWork(msg *Message) {
 
 func (b *Broker) Start() {
 	if b == nil {
-		log.Error("broker is null")
+		brokerLog.Error("broker is null")
 		return
 	}
+	go b.StartDispatcher()
 
 	//listen clinet over tcp
 	if b.config.Port != "" {
@@ -157,7 +149,7 @@ func StateMonitor() {
 func (b *Broker) StartWebsocketListening() {
 	path := b.config.WsPath
 	hp := ":" + b.config.WsPort
-	log.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path))
+	brokerLog.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path))
 	http.Handle(path, websocket.Handler(b.wsHandler))
 	var err error
 	if b.config.WsTLS {
@@ -166,7 +158,7 @@ func (b *Broker) StartWebsocketListening() {
 		err = http.ListenAndServe(hp, nil)
 	}
 	if err != nil {
-		log.Error("ListenAndServe:" + err.Error())
+		brokerLog.Error("ListenAndServe:" + err.Error())
 		return
 	}
 }
@@ -175,7 +167,7 @@ func (b *Broker) wsHandler(ws *websocket.Conn) {
 	// io.Copy(ws, ws)
 	atomic.AddUint64(&b.cid, 1)
 	ws.PayloadType = websocket.BinaryFrame
-	b.handleConnection(CLIENT, ws)
+	b.handleConnection(CLIENT, ws, b.cid)
 }
 
 func (b *Broker) StartClientListening(Tls bool) {
@@ -185,14 +177,14 @@ func (b *Broker) StartClientListening(Tls bool) {
 	if Tls {
 		hp = b.config.TlsHost + ":" + b.config.TlsPort
 		l, err = tls.Listen("tcp", hp, b.tlsConfig)
-		log.Info("Start TLS Listening client on ", zap.String("hp", hp))
+		brokerLog.Info("Start TLS Listening client on ", zap.String("hp", hp))
 	} else {
 		hp := b.config.Host + ":" + b.config.Port
 		l, err = net.Listen("tcp", hp)
-		log.Info("Start Listening client on ", zap.String("hp", hp))
+		brokerLog.Info("Start Listening client on ", zap.String("hp", hp))
 	}
 	if err != nil {
-		log.Error("Error listening on ", zap.Error(err))
+		brokerLog.Error("Error listening on ", zap.Error(err))
 		return
 	}
 	tmpDelay := 10 * ACCEPT_MIN_SLEEP
@@ -200,7 +192,7 @@ func (b *Broker) StartClientListening(Tls bool) {
 		conn, err := l.Accept()
 		if err != nil {
 			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				log.Error("Temporary Client Accept Error(%v), sleeping %dms",
+				brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
 					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
@@ -208,13 +200,13 @@ func (b *Broker) StartClientListening(Tls bool) {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				log.Error("Accept error: %v", zap.Error(err))
+				brokerLog.Error("Accept error: %v", zap.Error(err))
 			}
 			continue
 		}
 		tmpDelay = ACCEPT_MIN_SLEEP
 		atomic.AddUint64(&b.cid, 1)
-		go b.handleConnection(CLIENT, conn)
+		go b.handleConnection(CLIENT, conn, b.cid)
 
 	}
 }
@@ -227,7 +219,7 @@ func (b *Broker) Handshake(conn net.Conn) bool {
 
 	// Force handshake
 	if err := nc.Handshake(); err != nil {
-		log.Error("TLS handshake error, ", zap.Error(err))
+		brokerLog.Error("TLS handshake error, ", zap.Error(err))
 		return false
 	}
 	nc.SetReadDeadline(time.Time{})
@@ -243,27 +235,28 @@ func TlsTimeout(conn *tls.Conn) {
 	}
 	cs := nc.ConnectionState()
 	if !cs.HandshakeComplete {
-		log.Error("TLS handshake timeout")
+		brokerLog.Error("TLS handshake timeout")
 		nc.Close()
 	}
 }
 
 func (b *Broker) StartClusterListening() {
 	var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port
-	log.Info("Start Listening cluster on ", zap.String("hp", hp))
+	brokerLog.Info("Start Listening cluster on ", zap.String("hp", hp))
 
 	l, e := net.Listen("tcp", hp)
 	if e != nil {
-		log.Error("Error listening on ", zap.Error(e))
+		brokerLog.Error("Error listening on ", zap.Error(e))
 		return
 	}
 
+	var idx uint64 = 0
 	tmpDelay := 10 * ACCEPT_MIN_SLEEP
 	for {
 		conn, err := l.Accept()
 		if err != nil {
 			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				log.Error("Temporary Client Accept Error(%v), sleeping %dms",
+				brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
 					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
@@ -271,30 +264,30 @@ func (b *Broker) StartClusterListening() {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				log.Error("Accept error: %v", zap.Error(err))
+				brokerLog.Error("Accept error: %v", zap.Error(err))
 			}
 			continue
 		}
 		tmpDelay = ACCEPT_MIN_SLEEP
 
-		go b.handleConnection(ROUTER, conn)
+		go b.handleConnection(ROUTER, conn, idx)
 	}
 }
 
-func (b *Broker) handleConnection(typ int, conn net.Conn) {
+func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	//process connect packet
 	packet, err := packets.ReadPacket(conn)
 	if err != nil {
-		log.Error("read connect packet error: ", zap.Error(err))
+		brokerLog.Error("read connect packet error: ", zap.Error(err))
 		return
 	}
 	if packet == nil {
-		log.Error("received nil packet")
+		brokerLog.Error("received nil packet")
 		return
 	}
 	msg, ok := packet.(*packets.ConnectPacket)
 	if !ok {
-		log.Error("received msg that was not Connect")
+		brokerLog.Error("received msg that was not Connect")
 		return
 	}
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
@@ -302,7 +295,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	connack.SessionPresent = msg.CleanSession
 	err = connack.Write(conn)
 	if err != nil {
-		log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
+		brokerLog.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
 		return
 	}
 
@@ -342,7 +335,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	case CLIENT:
 		old, exist = b.clients.Load(cid)
 		if exist {
-			log.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
+			brokerLog.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
 			ol, ok := old.(*client)
 			if ok {
 				ol.Close()
@@ -352,7 +345,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	case ROUTER:
 		old, exist = b.routes.Load(cid)
 		if exist {
-			log.Warn("router exist, close old...")
+			brokerLog.Warn("router exist, close old...")
 			ol, ok := old.(*client)
 			if ok {
 				ol.Close()
@@ -361,9 +354,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 		b.routes.Store(cid, c)
 	}
 
-	// mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum]
-
-	c.readLoop()
+	c.readLoop(b.messagePool)
 }
 
 func (b *Broker) ConnectToDiscovery() {
@@ -373,8 +364,8 @@ func (b *Broker) ConnectToDiscovery() {
 	for {
 		conn, err = net.Dial("tcp", b.config.Router)
 		if err != nil {
-			log.Error("Error trying to connect to route: ", zap.Error(err))
-			log.Debug("Connect to route timeout ,retry...")
+			brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
+			brokerLog.Debug("Connect to route timeout ,retry...")
 
 			if 0 == tempDelay {
 				tempDelay = 1 * time.Second
@@ -390,7 +381,7 @@ func (b *Broker) ConnectToDiscovery() {
 		}
 		break
 	}
-	log.Debug("connect to router success :", zap.String("Router", b.config.Router))
+	brokerLog.Debug("connect to router success :", zap.String("Router", b.config.Router))
 
 	cid := b.id
 	info := info{
@@ -410,7 +401,7 @@ func (b *Broker) ConnectToDiscovery() {
 	c.SendConnect()
 	c.SendInfo()
 
-	go c.readLoop()
+	go c.readLoop(b.clusterPool)
 	go c.StartPing()
 }
 
@@ -418,7 +409,7 @@ func (b *Broker) processClusterInfo() {
 	for {
 		msg, ok := <-b.clusterPool
 		if !ok {
-			log.Error("read message from cluster channel error")
+			brokerLog.Error("read message from cluster channel error")
 			return
 		}
 		ProcessMessage(msg)
@@ -440,13 +431,13 @@ func (b *Broker) connectRouter(id, addr string) {
 
 		conn, err = net.Dial("tcp", addr)
 		if err != nil {
-			log.Error("Error trying to connect to route: ", zap.Error(err))
+			brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
 
 			if retryTimes > 50 {
 				return
 			}
 
-			log.Debug("Connect to route timeout ,retry...")
+			brokerLog.Debug("Connect to route timeout ,retry...")
 
 			if 0 == timeDelay {
 				timeDelay = 1 * time.Second
@@ -486,8 +477,7 @@ func (b *Broker) connectRouter(id, addr string) {
 
 	c.SendConnect()
 
-	// mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum]
-	go c.readLoop()
+	go c.readLoop(b.messagePool)
 	go c.StartPing()
 
 }
@@ -546,7 +536,7 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 	if len(subInfo.Topics) > 0 {
 		err := c.WriterPacket(subInfo)
 		if err != nil {
-			log.Error("Send localsubs To Router error :", zap.Error(err))
+			brokerLog.Error("Send localsubs To Router error :", zap.Error(err))
 		}
 	}
 }
@@ -563,7 +553,7 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke
 		return true
 
 	})
-	// log.Info("BroadcastInfoMessage success ")
+	// brokerLog.Info("BroadcastInfoMessage success ")
 }
 
 func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
@@ -575,7 +565,7 @@ func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
 		}
 		return true
 	})
-	// log.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
+	// brokerLog.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
 }
 
 func (b *Broker) removeClient(c *client) {
@@ -589,7 +579,7 @@ func (b *Broker) removeClient(c *client) {
 	case REMOTE:
 		b.remotes.Delete(clientId)
 	}
-	// log.Info("delete client ,", clientId)
+	// brokerLog.Info("delete client ,", clientId)
 }
 
 func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
@@ -603,7 +593,7 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 		if sub != nil {
 			err := sub.client.WriterPacket(packet)
 			if err != nil {
-				log.Error("process message for psub error,  ", zap.Error(err))
+				brokerLog.Error("process message for psub error,  ", zap.Error(err))
 			}
 		}
 	}

broker/client.go

@@ -86,11 +86,8 @@ func (c *client) init() {
 	c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0]
 }
 
-func (c *client) keepAlive(ch chan int) {
+func (c *client) keepAlive(ch chan int, mpool chan *Message) {
 	defer close(ch)
-
-	b := c.broker
-
 	keepalive := time.Duration(c.info.keepalive*3/2) * time.Second
 	timer := time.NewTimer(keepalive)
 
@@ -103,11 +100,9 @@ func (c *client) keepAlive(ch chan int) {
 				timer.Reset(keepalive)
 				continue
 			}
-			log.Error("Client exceeded timeout, disconnecting. ", zap.String("ClientID", c.info.clientID), zap.Uint16("keepalive", c.info.keepalive))
-
+			brokerLog.Error("Client exceeded timeout, disconnecting. ", zap.String("ClientID", c.info.clientID), zap.Uint16("keepalive", c.info.keepalive))
 			msg := &Message{client: c, packet: DisconnectdPacket}
-			b.SubmitWork(msg)
-
+			mpool <- msg
 			timer.Stop()
 			return
 		case _, ok := <-c.closed:
@@ -118,20 +113,19 @@ func (c *client) keepAlive(ch chan int) {
 	}
 }
 
-func (c *client) readLoop() {
+func (c *client) readLoop(mpool chan *Message) {
 	nc := c.conn
-	b := c.broker
-	if nc == nil || b == nil {
+	if nc == nil || mpool == nil {
 		return
 	}
 
 	ch := make(chan int, 1000)
-	go c.keepAlive(ch)
+	go c.keepAlive(ch, mpool)
 
 	for {
 		packet, err := packets.ReadPacket(nc)
 		if err != nil {
-			log.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+			brokerLog.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			break
 		}
 		// keepalive channel
@@ -141,11 +135,11 @@ func (c *client) readLoop() {
 			client: c,
 			packet: packet,
 		}
-		b.SubmitWork(msg)
+		mpool <- msg
 	}
 
 	msg := &Message{client: c, packet: DisconnectdPacket}
-	b.SubmitWork(msg)
+	mpool <- msg
 }
 
 func ProcessMessage(msg *Message) {
@@ -155,7 +149,7 @@ func ProcessMessage(msg *Message) {
 		return
 	}
 
-	log.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID))
+	brokerLog.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID))
 	switch ca.(type) {
 	case *packets.ConnackPacket:
 	case *packets.ConnectPacket:
@@ -180,7 +174,7 @@ func ProcessMessage(msg *Message) {
 	case *packets.DisconnectPacket:
 		c.Close()
 	default:
-		log.Info("Recv Unknow message.......", zap.String("ClientID", c.info.clientID))
+		brokerLog.Info("Recv Unknow message.......", zap.String("ClientID", c.info.clientID))
 	}
 }
 
@@ -196,7 +190,7 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) {
 	}
 
 	if !c.CheckTopicAuth(PUB, topic) {
-		log.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
+		brokerLog.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
 		return
 	}
 
@@ -207,21 +201,21 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) {
 		puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
 		puback.MessageID = packet.MessageID
 		if err := c.WriterPacket(puback); err != nil {
-			log.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+			brokerLog.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			return
 		}
 		c.ProcessPublishMessage(packet)
 	case QosExactlyOnce:
 		return
 	default:
-		log.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID))
+		brokerLog.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID))
 		return
 	}
 	if packet.Retain {
 		if b := c.broker; b != nil {
 			err := b.rl.Insert(topic, packet)
 			if err != nil {
-				log.Error("Insert Retain Message error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+				brokerLog.Error("Insert Retain Message error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			}
 		}
 	}
@@ -241,7 +235,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 	topic := packet.TopicName
 
 	r := b.sl.Match(topic)
-	// log.Info("psubs num: ", len(r.psubs))
+	// brokerLog.Info("psubs num: ", len(r.psubs))
 	if len(r.qsubs) == 0 && len(r.psubs) == 0 {
 		return
 	}
@@ -255,7 +249,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 		if sub != nil {
 			err := sub.client.WriterPacket(packet)
 			if err != nil {
-				log.Error("process message for psub error,  ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+				brokerLog.Error("process message for psub error,  ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			}
 		}
 	}
@@ -265,7 +259,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 	t := "$queue/" + topic
 	cnt, exist := b.queues[t]
 	if exist {
-		// log.Info("queue index : ", cnt)
+		// brokerLog.Info("queue index : ", cnt)
 		for _, sub := range r.qsubs {
 			if sub.client.typ == ROUTER {
 				if typ != CLIENT {
@@ -281,7 +275,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 				if sub != nil {
 					err := sub.client.WriterPacket(packet)
 					if err != nil {
-						log.Error("send publish error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+						brokerLog.Error("send publish error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 					}
 				}
 
@@ -335,7 +329,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 		t := topic
 		//check topic auth for client
 		if !c.CheckTopicAuth(SUB, topic) {
-			log.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
+			brokerLog.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
 			retcodes = append(retcodes, QosFailure)
 			continue
 		}
@@ -382,7 +376,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 		}
 		err := b.sl.Insert(sub)
 		if err != nil {
-			log.Error("Insert subscription error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+			brokerLog.Error("Insert subscription error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			retcodes = append(retcodes, QosFailure)
 		} else {
 			retcodes = append(retcodes, qoss[i])
@@ -392,7 +386,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 
 	err := c.WriterPacket(suback)
 	if err != nil {
-		log.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+		brokerLog.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 		return
 	}
 	//broadcast subscribe message
@@ -404,7 +398,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 	for _, t := range topics {
 		packets := b.rl.Match(t)
 		for _, packet := range packets {
-			log.Info("process retain  message: ", zap.Any("packet", packet), zap.String("ClientID", c.info.clientID))
+			brokerLog.Info("process retain  message: ", zap.Any("packet", packet), zap.String("ClientID", c.info.clientID))
 			if packet != nil {
 				c.WriterPacket(packet)
 			}
@@ -451,7 +445,7 @@ func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
 
 	err := c.WriterPacket(unsuback)
 	if err != nil {
-		log.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+		brokerLog.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 		return
 	}
 	// //process ubsubscribe message
@@ -480,7 +474,7 @@ func (c *client) ProcessPing() {
 	resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
 	err := c.WriterPacket(resp)
 	if err != nil {
-		log.Error("send PingResponse error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+		brokerLog.Error("send PingResponse error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 		return
 	}
 }
@@ -511,7 +505,7 @@ func (c *client) Close() {
 		for _, sub := range subs {
 			err := b.sl.Remove(sub)
 			if err != nil {
-				log.Error("closed client but remove sublist error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
+				brokerLog.Error("closed client but remove sublist error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			}
 		}
 		if c.typ == CLIENT {
@@ -533,7 +527,7 @@ func (c *client) Close() {
 }
 
 func (c *client) WriterPacket(packet packets.ControlPacket) error {
-	if c == nil || packet == nil {
+	if packet == nil {
 		return nil
 	}
 

broker/config.go

@@ -9,10 +9,11 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"github.com/fhmq/hmq/logger"
-	"go.uber.org/zap"
 	"io/ioutil"
 	"os"
+
+	"github.com/fhmq/hmq/logger"
+	"go.uber.org/zap"
 )
 
 type Config struct {
@@ -51,10 +52,6 @@ var DefaultConfig *Config = &Config{
 	Acl:    false,
 }
 
-var (
-	log *zap.Logger
-)
-
 func showHelp() {
 	fmt.Printf("%s\n", usageStr)
 	os.Exit(0)
@@ -108,7 +105,7 @@ func ConfigureConfig(args []string) (*Config, error) {
 	})
 
 	logger.InitLogger(config.Debug)
-	log = logger.Get().Named("Broker")
+	brokerLog = logger.Get().Named("Broker")
 
 	if configFile != "" {
 		tmpConfig, e := LoadConfig(configFile)
@@ -131,15 +128,15 @@ func LoadConfig(filename string) (*Config, error) {
 
 	content, err := ioutil.ReadFile(filename)
 	if err != nil {
-		log.Error("Read config file error: ", zap.Error(err))
+		brokerLog.Error("Read config file error: ", zap.Error(err))
 		return nil, err
 	}
-	// log.Info(string(content))
+	// brokerLog.Info(string(content))
 
 	var config Config
 	err = json.Unmarshal(content, &config)
 	if err != nil {
-		log.Error("Unmarshal config file error: ", zap.Error(err))
+		brokerLog.Error("Unmarshal config file error: ", zap.Error(err))
 		return nil, err
 	}
 
@@ -171,7 +168,7 @@ func (config *Config) check() error {
 
 	if config.TlsPort != "" {
 		if config.TlsInfo.CertFile == "" || config.TlsInfo.KeyFile == "" {
-			log.Error("tls config error, no cert or key file.")
+			brokerLog.Error("tls config error, no cert or key file.")
 			return errors.New("tls config error, no cert or key file.")
 		}
 		if config.TlsHost == "" {

broker/info.go

@@ -4,10 +4,12 @@ package broker
 
 import (
 	"fmt"
-	simplejson "github.com/bitly/go-simplejson"
+	"time"
+
 	"github.com/eclipse/paho.mqtt.golang/packets"
 	"go.uber.org/zap"
-	"time"
+
+	simplejson "github.com/bitly/go-simplejson"
 )
 
 func (c *client) SendInfo() {
@@ -19,7 +21,7 @@ func (c *client) SendInfo() {
 	infoMsg := NewInfo(c.broker.id, url, false)
 	err := c.WriterPacket(infoMsg)
 	if err != nil {
-		log.Error("send info message error, ", zap.Error(err))
+		brokerLog.Error("send info message error, ", zap.Error(err))
 		return
 	}
 }
@@ -32,7 +34,7 @@ func (c *client) StartPing() {
 		case <-timeTicker.C:
 			err := c.WriterPacket(ping)
 			if err != nil {
-				log.Error("ping error: ", zap.Error(err))
+				brokerLog.Error("ping error: ", zap.Error(err))
 				c.Close()
 			}
 		case _, ok := <-c.closed:
@@ -55,10 +57,10 @@ func (c *client) SendConnect() {
 	m.Keepalive = uint16(60)
 	err := c.WriterPacket(m)
 	if err != nil {
-		log.Error("send connect message error, ", zap.Error(err))
+		brokerLog.Error("send connect message error, ", zap.Error(err))
 		return
 	}
-	log.Info("send connect success")
+	brokerLog.Info("send connect success")
 }
 
 func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
@@ -67,7 +69,7 @@ func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
 	pub.TopicName = BrokerInfoTopic
 	pub.Retain = false
 	info := fmt.Sprintf(`{"brokerID":"%s","brokerUrl":"%s"}`, sid, url)
-	// log.Info("new info", string(info))
+	// brokerLog.Info("new info", string(info))
 	pub.Payload = []byte(info)
 	return pub
 }
@@ -79,17 +81,17 @@ func (c *client) ProcessInfo(packet *packets.PublishPacket) {
 		return
 	}
 
-	log.Info("recv remoteInfo: ", zap.String("payload", string(packet.Payload)))
+	brokerLog.Info("recv remoteInfo: ", zap.String("payload", string(packet.Payload)))
 
 	js, err := simplejson.NewJson(packet.Payload)
 	if err != nil {
-		log.Warn("parse info message err", zap.Error(err))
+		brokerLog.Warn("parse info message err", zap.Error(err))
 		return
 	}
 
 	routes, err := js.Get("data").Map()
 	if routes == nil {
-		log.Error("receive info message error, ", zap.Error(err))
+		brokerLog.Error("receive info message error, ", zap.Error(err))
 		return
 	}
 

broker/retain.go

@@ -1,8 +1,9 @@
 package broker
 
 import (
-	"github.com/eclipse/paho.mqtt.golang/packets"
 	"sync"
+
+	"github.com/eclipse/paho.mqtt.golang/packets"
 )
 
 type RetainList struct {
@@ -38,7 +39,7 @@ func (r *RetainList) Insert(topic string, buf *packets.PublishPacket) error {
 	if err != nil {
 		return err
 	}
-	// log.Info("insert tokens:", tokens)
+	// brokerLog.Info("insert tokens:", tokens)
 	r.Lock()
 
 	l := r.root
@@ -71,7 +72,7 @@ func (r *RetainList) Match(topic string) []*packets.PublishPacket {
 	l := r.root
 	matchRLevel(l, tokens, results)
 	r.Unlock()
-	// log.Info("results: ", results)
+	// brokerLog.Info("results: ", results)
 	return results.msg
 
 }
@@ -81,7 +82,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) {
 		if l == nil {
 			return
 		}
-		// log.Info("l info :", l.nodes)
+		// brokerLog.Info("l info :", l.nodes)
 		if t == "#" {
 			for _, n := range l.nodes {
 				n.GetAll(results)
@@ -110,7 +111,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) {
 }
 
 func (r *rnode) GetAll(results *RetainResult) {
-	// log.Info("node 's message: ", string(r.msg))
+	// brokerLog.Info("node 's message: ", string(r.msg))
 	if r.msg != nil {
 		results.msg = append(results.msg, r.msg)
 	}

broker/sublist.go

@@ -4,8 +4,9 @@ package broker
 
 import (
 	"errors"
-	"go.uber.org/zap"
 	"sync"
+
+	"go.uber.org/zap"
 )
 
 // A result structure better optimized for queue subs.
@@ -210,7 +211,7 @@ func (s *Sublist) Match(topic string) *SublistResult {
 
 	tokens, err := PublishTopicCheckAndSpilt(topic)
 	if err != nil {
-		log.Error("\tserver/sublist.go: ", zap.Error(err))
+		brokerLog.Error("\tserver/sublist.go: ", zap.Error(err))
 		return nil
 	}
 

main.go

@@ -8,10 +8,11 @@ package main
 
 import (
 	"fmt"
-	"github.com/fhmq/hmq/broker"
 	"os"
 	"os/signal"
 	"runtime"
+
+	"github.com/fhmq/hmq/broker"
 )
 
 func main() {