modify Dockerfile

Dockerfile

@@ -1,6 +1,5 @@
 FROM alpine
 COPY hmq /
-COPY hmq.config /
 COPY ssl /ssl
 COPY conf /conf
 

_config.yml

@@ -1 +0,0 @@
-theme: jekyll-theme-slate

broker/broker.go

@@ -21,6 +21,7 @@ import (
 type Broker struct {
 	id        string
 	cid       uint64
+	mu        sync.Mutex
 	config    *Config
 	tlsConfig *tls.Config
 	AclConfig *acl.ACLConfig
@@ -65,35 +66,47 @@ func (b *Broker) Start() {
 		log.Error("broker is null")
 		return
 	}
+	StartDispatcher()
+
+	//listen clinet over tcp
 	if b.config.Port != "" {
 		go b.StartClientListening(false)
 	}
+
+	//listen for cluster
 	if b.config.Cluster.Port != "" {
 		go b.StartClusterListening()
 	}
+
+	//listen for websocket
 	if b.config.WsPort != "" {
 		go b.StartWebsocketListening()
 	}
+
+	//listen client over tls
 	if b.config.TlsPort != "" {
 		go b.StartClientListening(true)
 	}
+
+	//connect on other node in cluster
 	if len(b.config.Cluster.Routes) > 0 {
 		b.ConnectToRouters()
 	}
+
+	//system montior
 	go StateMonitor()
 
 }
 
 func StateMonitor() {
 	v, _ := mem.VirtualMemory()
-	timeSticker := time.NewTicker(time.Second * 5)
+	timeSticker := time.NewTicker(time.Second * 30)
 	for {
 		select {
 		case <-timeSticker.C:
-			if v.UsedPercent > 0.8 {
+			if v.UsedPercent > 75 {
 				debug.FreeOSMemory()
 			}
-			// fmt.Printf("Total: %v, Free:%v, UsedPercent:%f%%\n", v.Total, v.Free, v.UsedPercent)
 		}
 	}
 }
@@ -223,12 +236,6 @@ func (b *Broker) StartClusterListening() {
 		tmpDelay = ACCEPT_MIN_SLEEP
 
 		go b.handleConnection(ROUTER, conn, idx)
-		if idx == 1 {
-			idx = 0
-		} else {
-			idx = idx + 1
-		}
-
 	}
 }
 
@@ -253,7 +260,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	connack.SessionPresent = msg.CleanSession
 	err = connack.Write(conn)
 	if err != nil {
-		log.Error("send connack error, ", err)
+		log.Error("send connack error, ", err, " clientID = ", msg.ClientIdentifier)
 		return
 	}
 
@@ -281,6 +288,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 		conn:   conn,
 		info:   info,
 	}
+
 	c.init()
 
 	cid := c.info.clientID
@@ -292,28 +300,42 @@ 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...", " clientID = ", c.info.clientID)
+			ol, ok := old.(*client)
+			if ok {
+				msg := &Message{client: c, packet: DisconnectdPacket}
+				ol.mp.queue <- msg
+			}
+		}
 		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 {
+	go c.readLoop()
-		log.Warn("client or routers exist, close old...")
+	if typ == ROUTER {
-		ol, ok := old.(*client)
+		c.SendInfo()
-		if ok {
+		c.StartPing()
-			ol.Close()
-		}
 	}
-	c.readLoop(msgPool)
 }
 
 func (b *Broker) ConnectToRouters() {
-	for i := 0; i < len(b.config.Cluster.Routes); i++ {
+	for _, v := range b.config.Cluster.Routes {
-		url := b.config.Cluster.Routes[i]
+		go b.connectRouter(v, "")
-		go b.connectRouter(url, "")
 	}
 }
 
@@ -324,32 +346,41 @@ func (b *Broker) connectRouter(url, remoteID string) {
 		conn, err = net.Dial("tcp", url)
 		if err != nil {
 			log.Error("Error trying to connect to route: ", err)
-			select {
+			log.Debug("Connect to route timeout ,retry...")
-			case <-time.After(DEFAULT_ROUTE_CONNECT):
+			time.Sleep(5 * time.Second)
-				log.Debug("Connect to route timeout ,retry...")
+			continue
-				continue
-			}
 		}
+		break
 	}
-	route := &route{
+	route := route{
 		remoteID:  remoteID,
-		remoteUrl: url,
+		remoteUrl: conn.RemoteAddr().String(),
 	}
 	cid := GenUniqueId()
+
 	info := info{
-		clientID: cid,
+		clientID:  cid,
+		keepalive: 60,
 	}
+
 	c := &client{
-		typ:   REMOTE,
+		broker: b,
-		conn:  conn,
+		typ:    REMOTE,
-		route: route,
+		conn:   conn,
-		info:  info,
+		route:  route,
+		info:   info,
 	}
 	c.init()
 	b.remotes.Store(cid, c)
+
+	c.mp = MSGPool[(MessagePoolNum + 1)].GetPool()
+
 	c.SendConnect()
 	c.SendInfo()
-	c.StartPing()
+
+	go c.readLoop()
+	go c.StartPing()
+
 }
 
 func (b *Broker) CheckRemoteExist(remoteID, url string) bool {
@@ -358,9 +389,7 @@ func (b *Broker) CheckRemoteExist(remoteID, url string) bool {
 		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
 				return false
 			}
@@ -383,14 +412,16 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 		}
 		return true
 	})
-	err := c.WriterPacket(subInfo)
+	if len(subInfo.Topics) > 0 {
-	if err != nil {
+		err := c.WriterPacket(subInfo)
-		log.Error("Send localsubs To Router error :", err)
+		if err != nil {
+			log.Error("Send localsubs To Router error :", err)
+		}
 	}
 }
 
 func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacket) {
-	b.remotes.Range(func(key, value interface{}) bool {
+	b.routes.Range(func(key, value interface{}) bool {
 		r, ok := value.(*client)
 		if ok {
 			if r.route.remoteID == remoteID {
@@ -405,7 +436,8 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke
 }
 
 func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
-	b.remotes.Range(func(key, value interface{}) bool {
+
+	b.routes.Range(func(key, value interface{}) bool {
 		r, ok := value.(*client)
 		if ok {
 			r.WriterPacket(packet)
@@ -432,7 +464,6 @@ func (b *Broker) removeClient(c *client) {
 func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 	topic := packet.TopicName
 	r := b.sl.Match(topic)
-	// log.Info("psubs num: ", len(r.psubs))
 	if len(r.psubs) == 0 {
 		return
 	}
@@ -449,14 +480,12 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 
 func (b *Broker) BroadcastUnSubscribe(subs map[string]*subscription) {
 
-	ubsub := packets.NewControlPacket(packets.Unsubscribe).(*packets.UnsubscribePacket)
+	unsub := packets.NewControlPacket(packets.Unsubscribe).(*packets.UnsubscribePacket)
 	for topic, _ := range subs {
-		// topic := sub.topic
+		unsub.Topics = append(unsub.Topics, topic)
-		// if sub.queue {
-		// 	topic = "$queue/" + sub.topic
-		// }
-		ubsub.Topics = append(ubsub.Topics, topic)
 	}
-	b.BroadcastSubOrUnsubMessage(ubsub)
 
+	if len(unsub.Topics) > 0 {
+		b.BroadcastSubOrUnsubMessage(unsub)
+	}
 }

broker/client.go

@@ -32,9 +32,11 @@ type client struct {
 	broker *Broker
 	conn   net.Conn
 	info   info
-	route  *route
+	route  route
 	status int
+	closed chan int
 	smu    sync.RWMutex
+	mp     *MessagePool
 	subs   map[string]*subscription
 	rsubs  map[string]*subInfo
 }
@@ -74,47 +76,65 @@ func (c *client) init() {
 	c.smu.Lock()
 	defer c.smu.Unlock()
 	c.status = Connected
-	typ := c.typ
+	c.closed = make(chan int, 1)
-	if typ == ROUTER {
+
-		c.rsubs = make(map[string]*subInfo)
+	c.rsubs = make(map[string]*subInfo)
-	} else if typ == CLIENT {
+	c.subs = make(map[string]*subscription, 10)
-		c.subs = make(map[string]*subscription, 10)
-	}
 	c.info.localIP = strings.Split(c.conn.LocalAddr().String(), ":")[0]
 	c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0]
 }
 
-func (c *client) readLoop(msgPool *MessagePool) {
+func (c *client) keepAlive(ch chan int) {
+	defer close(ch)
+	keepalive := time.Duration(c.info.keepalive*3/2) * time.Second
+	timer := time.NewTimer(keepalive)
+	msgPool := c.mp
+
+	for {
+		select {
+		case <-ch:
+			timer.Reset(keepalive)
+		case <-timer.C:
+			log.Error("Client exceeded timeout, disconnecting. clientID = ", c.info.clientID, " keepalive = ", c.info.keepalive)
+			msg := &Message{client: c, packet: DisconnectdPacket}
+			msgPool.queue <- msg
+			timer.Stop()
+			return
+		case _, ok := <-c.closed:
+			if !ok {
+				return
+			}
+		}
+	}
+}
+
+func (c *client) readLoop() {
 	nc := c.conn
+	msgPool := c.mp
 	if nc == nil || msgPool == nil {
 		return
 	}
 
-	lastIn := uint16(time.Now().Unix())
+	ch := make(chan int, 1000)
-	var nowTime uint16
+	go c.keepAlive(ch)
+
 	for {
-		nowTime = uint16(time.Now().Unix())
-		if 0 != c.info.keepalive && nowTime-lastIn > c.info.keepalive*3/2 {
-			log.Errorf("Client %s has exceeded timeout, disconnecting.\n", c.info.clientID)
-			msg := &Message{client: c, packet: DisconnectdPacket}
-			msgPool.queue <- msg
-			return
-		}
 		packet, err := packets.ReadPacket(nc)
 		if err != nil {
-			log.Error("read packet error: ", err)
+			log.Error("read packet error: ", err, " clientID = ", c.info.clientID)
-			msg := &Message{client: c, packet: DisconnectdPacket}
+			break
-			msgPool.queue <- msg
-			return
 		}
-		// log.Info("recv buf: ", packet)
+
-		lastIn = uint16(time.Now().Unix())
+		ch <- 1
+
 		msg := &Message{
 			client: c,
 			packet: packet,
 		}
 		msgPool.queue <- msg
 	}
+	msg := &Message{client: c, packet: DisconnectdPacket}
+	msgPool.queue <- msg
 	msgPool.Reduce()
 }
 
@@ -124,45 +144,33 @@ func ProcessMessage(msg *Message) {
 	if ca == nil {
 		return
 	}
+	log.Debug("Recv message: ", ca.String(), " clientID = ", c.info.clientID)
 	switch ca.(type) {
 	case *packets.ConnackPacket:
-		// log.Info("Recv conack message..........")
+
 	case *packets.ConnectPacket:
-		// log.Info("Recv connect message..........")
 	case *packets.PublishPacket:
-		// log.Info("Recv publish message..........")
 		packet := ca.(*packets.PublishPacket)
 		c.ProcessPublish(packet)
 	case *packets.PubackPacket:
-		//log.Info("Recv publish  ack message..........")
 	case *packets.PubrecPacket:
-		//log.Info("Recv publish rec message..........")
 	case *packets.PubrelPacket:
-		//log.Info("Recv publish rel message..........")
 	case *packets.PubcompPacket:
-		//log.Info("Recv publish  ack message..........")
 	case *packets.SubscribePacket:
-		// log.Info("Recv subscribe message.....")
 		packet := ca.(*packets.SubscribePacket)
 		c.ProcessSubscribe(packet)
 	case *packets.SubackPacket:
-		// log.Info("Recv suback message.....")
 	case *packets.UnsubscribePacket:
-		// log.Info("Recv unsubscribe message.....")
 		packet := ca.(*packets.UnsubscribePacket)
 		c.ProcessUnSubscribe(packet)
 	case *packets.UnsubackPacket:
-		//log.Info("Recv unsuback message.....")
 	case *packets.PingreqPacket:
-		// log.Info("Recv PINGREQ message..........")
 		c.ProcessPing()
 	case *packets.PingrespPacket:
-		//log.Info("Recv PINGRESP message..........")
 	case *packets.DisconnectPacket:
-		// log.Info("Recv DISCONNECT message.......")
 		c.Close()
 	default:
-		log.Info("Recv Unknow message.......")
+		log.Info("Recv Unknow message.......", " clientID = ", c.info.clientID)
 	}
 }
 
@@ -172,8 +180,13 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) {
 	}
 
 	topic := packet.TopicName
+	if topic == BrokerInfoTopic && c.typ != CLIENT {
+		c.ProcessInfo(packet)
+		return
+	}
+
 	if !c.CheckTopicAuth(PUB, topic) {
-		log.Error("Pub Topics Auth failed, ", topic)
+		log.Error("Pub Topics Auth failed, ", topic, " clientID = ", c.info.clientID)
 		return
 	}
 
@@ -184,21 +197,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, ", err)
+			log.Error("send puback error, ", err, " clientID = ", c.info.clientID)
 			return
 		}
 		c.ProcessPublishMessage(packet)
 	case QosExactlyOnce:
 		return
 	default:
-		log.Error("publish with unknown qos")
+		log.Error("publish with unknown qos", " 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: ", err)
+				log.Error("Insert Retain Message error: ", err, " clientID = ", c.info.clientID)
 			}
 		}
 	}
@@ -224,15 +237,15 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 	}
 
 	for _, sub := range r.psubs {
-		if sub.client.typ == ROUTER {
+		if sub.client.typ == REMOTE {
-			if typ == ROUTER {
+			if typ == REMOTE {
 				continue
 			}
 		}
 		if sub != nil {
 			err := sub.client.WriterPacket(packet)
 			if err != nil {
-				log.Error("process message for psub error,  ", err)
+				log.Error("process message for psub error,  ", err, " clientID = ", c.info.clientID)
 			}
 		}
 	}
@@ -244,8 +257,8 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 	if exist {
 		// log.Info("queue index : ", cnt)
 		for _, sub := range r.qsubs {
-			if sub.client.typ == ROUTER {
+			if sub.client.typ == REMOTE {
-				if c.typ == ROUTER {
+				if c.typ == REMOTE {
 					continue
 				}
 			}
@@ -258,7 +271,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 				if sub != nil {
 					err := sub.client.WriterPacket(packet)
 					if err != nil {
-						log.Error("send publish error, ", err)
+						log.Error("send publish error, ", err, " clientID = ", c.info.clientID)
 					}
 				}
 
@@ -312,7 +325,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: ", topic)
+			log.Error("Sub topic Auth failed: ", topic, " clientID = ", c.info.clientID)
 			retcodes = append(retcodes, QosFailure)
 			continue
 		}
@@ -346,7 +359,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 				retcodes = append(retcodes, qoss[i])
 				continue
 			}
-		case ROUTER:
+		case REMOTE:
 			if subinfo, exist := c.rsubs[topic]; !exist {
 				sinfo := &subInfo{sub: sub, num: 1}
 				c.rsubs[topic] = sinfo
@@ -359,7 +372,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 		}
 		err := b.sl.Insert(sub)
 		if err != nil {
-			log.Error("Insert subscription error: ", err)
+			log.Error("Insert subscription error: ", err, " clientID = ", c.info.clientID)
 			retcodes = append(retcodes, QosFailure)
 		} else {
 			retcodes = append(retcodes, qoss[i])
@@ -369,7 +382,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 
 	err := c.WriterPacket(suback)
 	if err != nil {
-		log.Error("send suback error, ", err)
+		log.Error("send suback error, ", err, " clientID = ", c.info.clientID)
 		return
 	}
 	//broadcast subscribe message
@@ -381,7 +394,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: ", packet)
+			log.Info("process retain  message: ", packet, " clientID = ", c.info.clientID)
 			if packet != nil {
 				c.WriterPacket(packet)
 			}
@@ -401,29 +414,25 @@ func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
 	topics := packet.Topics
 
 	for _, t := range topics {
-		var sub *subscription
+
-		ok := false
 		switch typ {
 		case CLIENT:
-			sub, ok = c.subs[t]
+			sub, ok := c.subs[t]
-		case ROUTER:
+			if ok {
+				c.unsubscribe(sub)
+			}
+		case REMOTE:
 			subinfo, ok := c.rsubs[t]
 			if ok {
 				subinfo.num = subinfo.num - 1
 				if subinfo.num < 1 {
-					sub = subinfo.sub
 					delete(c.rsubs, t)
+					c.unsubscribe(subinfo.sub)
 				} else {
 					c.rsubs[t] = subinfo
-					sub = nil
 				}
-			} else {
-				return
 			}
 		}
-		if ok {
-			go c.unsubscribe(sub)
-		}
 
 	}
 
@@ -432,7 +441,7 @@ func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
 
 	err := c.WriterPacket(unsuback)
 	if err != nil {
-		log.Error("send unsuback error, ", err)
+		log.Error("send unsuback error, ", err, " clientID = ", c.info.clientID)
 		return
 	}
 	// //process ubsubscribe message
@@ -461,19 +470,30 @@ func (c *client) ProcessPing() {
 	resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
 	err := c.WriterPacket(resp)
 	if err != nil {
-		log.Error("send PingResponse error, ", err)
+		log.Error("send PingResponse error, ", err, " clientID = ", c.info.clientID)
 		return
 	}
 }
 
 func (c *client) Close() {
 	c.smu.Lock()
+	if c.status == Disconnected {
+		c.smu.Unlock()
+		return
+	}
+	//wait for message complete
+	time.Sleep(1 * time.Second)
 	c.status = Disconnected
+
 	if c.conn != nil {
 		c.conn.Close()
 		c.conn = nil
 	}
+
 	c.smu.Unlock()
+
+	close(c.closed)
+
 	b := c.broker
 	subs := c.subs
 	if b != nil {
@@ -481,7 +501,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, ", err)
+				log.Error("closed client but remove sublist error, ", err, " clientID = ", c.info.clientID)
 			}
 		}
 		if c.typ == CLIENT {
@@ -490,10 +510,22 @@ func (c *client) Close() {
 		if c.info.willMsg != nil {
 			b.PublishMessage(c.info.willMsg)
 		}
+
+		//do reconnect
+		if c.typ == REMOTE {
+			localUrl := c.info.localIP + ":" + c.broker.config.Cluster.Port
+			if c.route.remoteUrl != localUrl {
+				b.connectRouter(c.route.remoteUrl, "")
+			}
+		}
 	}
 }
 
 func (c *client) WriterPacket(packet packets.ControlPacket) error {
+	if packet == nil {
+		return nil
+	}
+
 	c.mu.Lock()
 	err := packet.Write(c.conn)
 	c.mu.Unlock()

broker/config.go

@@ -11,7 +11,7 @@ import (
 )
 
 const (
-	CONFIGFILE = "hmq.config"
+	CONFIGFILE = "conf/hmq.config"
 )
 
 type Config struct {

broker/dispatcher.go

@@ -1,9 +1,5 @@
 package broker
 
-// const (
-// 	WorkNum = 4096
-// )
-
 var WorkNum int
 
 type Dispatcher struct {
@@ -31,7 +27,7 @@ func NewDispatcher() *Dispatcher {
 }
 
 func (d *Dispatcher) dispatch() {
-	for i := 0; i < MessagePoolNum; i++ {
+	for i := 0; i < (MessagePoolNum + 2); i++ {
 		go func(idx int) {
 			for {
 				select {

broker/info.go

@@ -22,7 +22,6 @@ func (c *client) SendInfo() {
 		log.Error("send info message error, ", err)
 		return
 	}
-	// log.Info("send info success")
 }
 
 func (c *client) StartPing() {
@@ -34,13 +33,19 @@ func (c *client) StartPing() {
 			err := c.WriterPacket(ping)
 			if err != nil {
 				log.Error("ping error: ", err)
+				c.Close()
+			}
+		case _, ok := <-c.closed:
+			if !ok {
+				return
 			}
 		}
 	}
 }
 
 func (c *client) SendConnect() {
-	if c.status == Disconnected {
+
+	if c.status != Connected {
 		return
 	}
 	m := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
@@ -53,7 +58,7 @@ func (c *client) SendConnect() {
 		log.Error("send connect message error, ", err)
 		return
 	}
-	// log.Info("send connet success")
+	log.Info("send connect success")
 }
 
 func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
@@ -98,17 +103,21 @@ func (c *client) ProcessInfo(packet *packets.PublishPacket) {
 		return
 	}
 
+	b.mu.Lock()
 	exist := b.CheckRemoteExist(rid, rurl)
 	if !exist {
-		go b.connectRouter(rurl, rid)
+		b.connectRouter(rurl, rid)
 	}
-	// log.Info("isforword: ", isForward)
+	b.mu.Unlock()
+
 	if !isForward {
-		route := &route{
+		if c.typ == ROUTER {
-			remoteUrl: rurl,
+			route := route{
-			remoteID:  rid,
+				remoteUrl: rurl,
+				remoteID:  rid,
+			}
+			c.route = route
 		}
-		c.route = route
 
 		go b.SendLocalSubsToRouter(c)
 		// log.Info("BroadcastInfoMessage starting... ")

broker/sublist.go

@@ -241,7 +241,6 @@ func (s *Sublist) Match(topic string) *SublistResult {
 	}
 
 	s.Unlock()
-	// log.Info("SublistResult: ", result)
 	return result
 }
 
@@ -294,7 +293,6 @@ func removeSubFromList(sub *subscription, sl []*subscription) ([]*subscription,
 			sl[i] = sl[last]
 			sl[last] = nil
 			sl = sl[:last]
-			// log.Info("removeSubFromList success")
 			return shrinkAsNeeded(sl), true
 		}
 	}

main.go

@@ -9,15 +9,31 @@ import (
 	log "github.com/cihub/seelog"
 )
 
+func init() {
+	testConfig := `
+<seelog type="sync">
+	<outputs formatid="main">
+		<console/>
+	</outputs>
+	<formats>
+		<format id="main" format="Time:%Date %Time%tfile:%File%tlevel:%LEVEL%t%Msg%n"/>
+	</formats>
+</seelog>`
+
+	logger, err := log.LoggerFromConfigAsBytes([]byte(testConfig))
+	if err != nil {
+		panic(err)
+	}
+	log.ReplaceLogger(logger)
+}
+
 func main() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
-	runtime.GC()
 	config, er := broker.LoadConfig()
 	if er != nil {
 		log.Error("Load Config file error: ", er)
 		return
 	}
-	broker.StartDispatcher()
 
 	b, err := broker.NewBroker(config)
 	if err != nil {