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,21 +66,34 @@ 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()
 
 }
@@ -93,7 +107,6 @@ func StateMonitor() {
 			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
-		}
-
 	}
 }
 
@@ -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
@@ -318,13 +326,16 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 		b.routes.Store(cid, c)
 	}
 
-	c.readLoop()
+	go c.readLoop()
+	if typ == ROUTER {
+		c.SendInfo()
+		c.StartPing()
+	}
 }
 
 func (b *Broker) ConnectToRouters() {
-	for i := 0; i < len(b.config.Cluster.Routes); i++ {
-		url := b.config.Cluster.Routes[i]
-		go b.connectRouter(url, "")
+	for _, v := range b.config.Cluster.Routes {
+		go b.connectRouter(v, "")
 	}
 }
 
@@ -335,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 {
-			case <-time.After(DEFAULT_ROUTE_CONNECT):
-				log.Debug("Connect to route timeout ,retry...")
-				continue
-			}
+			log.Debug("Connect to route timeout ,retry...")
+			time.Sleep(5 * time.Second)
+			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,
-		conn:  conn,
-		route: route,
-		info:  info,
+		broker: b,
+		typ:    REMOTE,
+		conn:   conn,
+		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 {
@@ -369,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
 			}
@@ -394,14 +412,16 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 		}
 		return true
 	})
-	err := c.WriterPacket(subInfo)
-	if err != nil {
-		log.Error("Send localsubs To Router error :", err)
+	if len(subInfo.Topics) > 0 {
+		err := c.WriterPacket(subInfo)
+		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 {
@@ -416,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)
@@ -443,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
 	}
@@ -460,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
-		// if sub.queue {
-		// 	topic = "$queue/" + sub.topic
-		// }
-		ubsub.Topics = append(ubsub.Topics, topic)
+		unsub.Topics = append(unsub.Topics, topic)
 	}
-	b.BroadcastSubOrUnsubMessage(ubsub)
 
+	if len(unsub.Topics) > 0 {
+		b.BroadcastSubOrUnsubMessage(unsub)
+	}
 }

broker/client.go

@@ -32,8 +32,9 @@ 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
@@ -75,16 +76,38 @@ 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)
-	} else if typ == CLIENT {
-		c.subs = make(map[string]*subscription, 10)
-	}
+	c.closed = make(chan int, 1)
+
+	c.rsubs = make(map[string]*subInfo)
+	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) 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
@@ -92,21 +115,18 @@ func (c *client) readLoop() {
 		return
 	}
 
-	lastIn := uint16(time.Now().Unix())
-	var nowTime uint16
+	ch := make(chan int, 1000)
+	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)
-			break
-		}
 		packet, err := packets.ReadPacket(nc)
 		if err != nil {
 			log.Error("read packet error: ", err, " clientID = ", c.info.clientID)
 			break
 		}
-		// log.Info("recv buf: ", packet)
-		lastIn = uint16(time.Now().Unix())
+
+		ch <- 1
+
 		msg := &Message{
 			client: c,
 			packet: packet,
@@ -160,6 +180,11 @@ 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, " clientID = ", c.info.clientID)
 		return
@@ -212,8 +237,8 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 	}
 
 	for _, sub := range r.psubs {
-		if sub.client.typ == ROUTER {
-			if typ == ROUTER {
+		if sub.client.typ == REMOTE {
+			if typ == REMOTE {
 				continue
 			}
 		}
@@ -232,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 c.typ == ROUTER {
+			if sub.client.typ == REMOTE {
+				if c.typ == REMOTE {
 					continue
 				}
 			}
@@ -334,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
@@ -389,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]
-		case ROUTER:
+			sub, ok := c.subs[t]
+			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)
-		}
 
 	}
 
@@ -455,21 +476,24 @@ func (c *client) ProcessPing() {
 }
 
 func (c *client) Close() {
+	c.smu.Lock()
 	if c.status == Disconnected {
+		c.smu.Unlock()
 		return
 	}
 	//wait for message complete
-	time.Sleep(time.Second)
-
-	c.smu.Lock()
+	time.Sleep(1 * time.Second)
 	c.status = Disconnected
-	c.smu.Unlock()
 
 	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 {
@@ -486,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{
-			remoteUrl: rurl,
-			remoteID:  rid,
+		if c.typ == ROUTER {
+			route := route{
+				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

@@ -34,7 +34,6 @@ func main() {
 		log.Error("Load Config file error: ", er)
 		return
 	}
-	broker.StartDispatcher()
 
 	b, err := broker.NewBroker(config)
 	if err != nil {