Merge branch 'master' of chowyu08.github.com:fhmq/hmq

.github/workflows/go.yml

@@ -1,23 +0,0 @@
-name: Go
-
-on: [push, pull_request]
-
-jobs:
-
-  build:
-    strategy:
-      matrix:
-        os: [ ubuntu-latest, windows-latest, macos-latest ]
-        goversion: [ 1.17 ]
-    
-    runs-on: ${{ matrix.os }}
-    steps:
-    - uses: actions/checkout@v2
-
-    - name: Set up Go
-      uses: actions/setup-go@v2
-      with:
-        go-version: ${{ matrix.goversion }}
-
-    - name: Build
-      run: go build -v ./...

.github/workflows/macos.yml

@@ -0,0 +1,18 @@
+name: MacOS build
+
+on: [push, pull_request]
+
+jobs:
+
+  build:
+    runs-on: macos-latest
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Set up Go
+      uses: actions/setup-go@v2
+      with:
+        go-version: 1.18
+
+    - name: Build
+      run: go build -v ./...

.github/workflows/ubuntu.yml

@@ -0,0 +1,18 @@
+name: Ubuntu build
+
+on: [push, pull_request]
+
+jobs:
+    
+    build:
+      runs-on: ubuntu-latest
+      steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: 1.18
+
+      - name: Build
+        run: go build -v ./...

.github/workflows/windows.yml

@@ -0,0 +1,18 @@
+name: Windows build
+
+on: [push, pull_request]
+
+jobs:
+    
+    build:
+      runs-on: windows-latest
+      steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: 1.18
+
+      - name: Build
+        run: go build -v ./...

.vscode/settings.json

@@ -1,11 +0,0 @@
-{
-    "go.lintFlags": [
-        "--disable=all",
-        "--enable=errcheck,varcheck,deadcode",
-        "--enable=varcheck",
-        "--enable=deadcode"
-    ],
-    "cSpell.words": [
-        "Authorised"
-    ]
-}

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.17 as builder
+FROM golang:1.18 as builder
 WORKDIR /go/src/github.com/fhmq/hmq
 COPY . .
 RUN CGO_ENABLED=0 go build -o hmq -a -ldflags '-extldflags "-static"' .
@@ -9,4 +9,4 @@ WORKDIR /
 COPY --from=builder /go/src/github.com/fhmq/hmq/hmq .
 EXPOSE 1883
 
-CMD ["/hmq"]
+ENTRYPOINT ["/hmq"]

README.md

@@ -1,3 +1,7 @@
+![build](https://img.shields.io/github/workflow/status/fhmq/hmq/Ubuntu%20build?label=Ubuntu&style=for-the-badge)
+![build](https://img.shields.io/github/workflow/status/fhmq/hmq/MacOS%20build?label=MacOS&style=for-the-badge)
+![build](https://img.shields.io/github/workflow/status/fhmq/hmq/Windows%20build?label=Windows&style=for-the-badge)
+
 Free and High Performance MQTT Broker 
 ============
 
@@ -135,7 +139,7 @@ Other Version Of Cluster Based On gRPC: [click here](https://github.com/fhmq/rhm
 
 ## Reference
 
-* Surgermq.(https://github.com/surgemq/surgemq)
+* Surgermq.(https://github.com/zentures/surgemq)
 
 ## Benchmark Tool
 

broker/bridge.go

@@ -5,11 +5,13 @@ import (
 	"go.uber.org/zap"
 )
 
-func (b *Broker) Publish(e *bridge.Elements) {
+func (b *Broker) Publish(e *bridge.Elements) bool {
 	if b.bridgeMQ != nil {
-		err := b.bridgeMQ.Publish(e)
+		cost, err := b.bridgeMQ.Publish(e)
 		if err != nil {
 			log.Error("send message to mq error.", zap.Error(err))
 		}
+		return cost
 	}
+	return false
 }

broker/broker.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	"golang.org/x/net/websocket"
 )
 
@@ -55,6 +56,37 @@ func newMessagePool() []chan *Message {
 	return pool
 }
 
+func getAdditionalLogFields(clientIdentifier string, conn net.Conn, additionalFields ...zapcore.Field) []zapcore.Field {
+	var wsConn *websocket.Conn = nil
+	var wsEnabled bool
+	result := []zapcore.Field{}
+
+	switch conn.(type) {
+	case *websocket.Conn:
+		wsEnabled = true
+		wsConn = conn.(*websocket.Conn)
+	case *net.TCPConn:
+		wsEnabled = false
+	}
+
+	// add optional fields
+	if len(additionalFields) > 0 {
+		result = append(result, additionalFields...)
+	}
+
+	// add client ID
+	result = append(result, zap.String("clientID", clientIdentifier))
+
+	// add remote connection address
+	if !wsEnabled && conn != nil && conn.RemoteAddr() != nil {
+		result = append(result, zap.Stringer("addr", conn.RemoteAddr()))
+	} else if wsEnabled && wsConn != nil && wsConn.Request() != nil {
+		result = append(result, zap.String("addr", wsConn.Request().RemoteAddr))
+	}
+
+	return result
+}
+
 func NewBroker(config *Config) (*Broker, error) {
 	if config == nil {
 		config = DefaultConfig
@@ -163,7 +195,7 @@ func (b *Broker) StartWebsocketListening() {
 		err = http.ListenAndServe(hp, mux)
 	}
 	if err != nil {
-		log.Error("ListenAndServe:" + err.Error())
+		log.Error("ListenAndServe" + err.Error())
 		return
 	}
 }
@@ -190,33 +222,39 @@ func (b *Broker) StartClientListening(Tls bool) {
 			l, err = net.Listen("tcp", hp)
 			log.Info("Start Listening client on ", zap.String("hp", hp))
 		}
-		if err != nil {
+
-			log.Error("Error listening on ", zap.Error(err))
+		if err == nil {
-			time.Sleep(1 * time.Second)
-		} else {
 			break // successfully listening
 		}
+
+		log.Error("Error listening on ", zap.Error(err))
+		time.Sleep(1 * time.Second)
 	}
+
 	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",
+				log.Error(
-					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
+					"Temporary Client Accept Error(%v), sleeping %dms",
+					zap.Error(ne),
+					zap.Duration("sleeping", tmpDelay/time.Millisecond),
+				)
+
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
 				if tmpDelay > ACCEPT_MAX_SLEEP {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				log.Error("Accept error: %v", zap.Error(err))
+				log.Error("Accept error", zap.Error(err))
 			}
 			continue
 		}
+
 		tmpDelay = ACCEPT_MIN_SLEEP
 		go b.handleConnection(CLIENT, conn)
-
 	}
 }
 
@@ -235,15 +273,19 @@ func (b *Broker) StartClusterListening() {
 		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",
+				log.Error(
-					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
+					"Temporary Client Accept Error(%v), sleeping %dms",
+					zap.Error(ne),
+					zap.Duration("sleeping", tmpDelay/time.Millisecond),
+				)
+
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
 				if tmpDelay > ACCEPT_MAX_SLEEP {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				log.Error("Accept error: %v", zap.Error(err))
+				log.Error("Accept error", zap.Error(err))
 			}
 			continue
 		}
@@ -269,7 +311,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	//process connect packet
 	packet, err := packets.ReadPacket(conn)
 	if err != nil {
-		log.Error("read connect packet error: ", zap.Error(err))
+		log.Error("read connect packet error", zap.Error(err))
 		conn.Close()
 		return
 	}
@@ -283,7 +325,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 		return
 	}
 
-	log.Info("read connect from ", zap.String("clientID", msg.ClientIdentifier))
+	log.Info("read connect from ", getAdditionalLogFields(msg.ClientIdentifier, conn)...)
 
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
 	connack.SessionPresent = msg.CleanSession
@@ -292,9 +334,8 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	if connack.ReturnCode != packets.Accepted {
 		func() {
 			defer conn.Close()
-			err = connack.Write(conn)
+			if err := connack.Write(conn); err != nil {
-			if err != nil {
+				log.Error("send connack error", getAdditionalLogFields(msg.ClientIdentifier, conn, zap.Error(err))...)
-				log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
 			}
 		}()
 		return
@@ -304,17 +345,15 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 		connack.ReturnCode = packets.ErrRefusedNotAuthorised
 		func() {
 			defer conn.Close()
-			err = connack.Write(conn)
+			if err := connack.Write(conn); err != nil {
-			if err != nil {
+				log.Error("send connack error", getAdditionalLogFields(msg.ClientIdentifier, conn, zap.Error(err))...)
-				log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
 			}
 		}()
 		return
 	}
 
-	err = connack.Write(conn)
+	if err := connack.Write(conn); err != nil {
-	if err != nil {
+		log.Error("send connack error", getAdditionalLogFields(msg.ClientIdentifier, conn, zap.Error(err))...)
-		log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
 		return
 	}
 
@@ -345,24 +384,22 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 
 	c.init()
 
-	err = b.getSession(c, msg, connack)
+	if err := b.getSession(c, msg, connack); err != nil {
-	if err != nil {
+		log.Error("get session error", getAdditionalLogFields(c.info.clientID, conn, zap.Error(err))...)
-		log.Error("get session error: ", zap.String("clientID", c.info.clientID))
 		return
 	}
 
 	cid := c.info.clientID
 
-	var exist bool
+	var exists bool
 	var old interface{}
 
 	switch typ {
 	case CLIENT:
-		old, exist = b.clients.Load(cid)
+		old, exists = b.clients.Load(cid)
-		if exist {
+		if exists {
-			log.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
+			if ol, ok := old.(*client); ok {
-			ol, ok := old.(*client)
+				log.Warn("client exists, close old client", getAdditionalLogFields(ol.info.clientID, ol.conn)...)
-			if ok {
 				ol.Close()
 			}
 		}
@@ -378,11 +415,10 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 			})
 		}
 	case ROUTER:
-		old, exist = b.routes.Load(cid)
+		old, exists = b.routes.Load(cid)
-		if exist {
+		if exists {
-			log.Warn("router exist, close old...")
+			if ol, ok := old.(*client); ok {
-			ol, ok := old.(*client)
+				log.Warn("router exists, close old router", getAdditionalLogFields(ol.info.clientID, ol.conn)...)
-			if ok {
 				ol.Close()
 			}
 		}
@@ -399,7 +435,7 @@ 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.Error("Error trying to connect to route", zap.Error(err))
 			log.Debug("Connect to route timeout, retry...")
 
 			if 0 == tempDelay {
@@ -416,7 +452,7 @@ func (b *Broker) ConnectToDiscovery() {
 		}
 		break
 	}
-	log.Debug("connect to router success :", zap.String("Router", b.config.Router))
+	log.Debug("connect to router success", zap.String("Router", b.config.Router))
 
 	cid := b.id
 	info := info{
@@ -466,7 +502,7 @@ 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))
+			log.Error("Error trying to connect to route", zap.Error(err))
 
 			if retryTimes > 50 {
 				return
@@ -540,19 +576,19 @@ func (b *Broker) checkNodeExist(id, url string) bool {
 }
 
 func (b *Broker) CheckRemoteExist(remoteID, url string) bool {
-	exist := false
+	exists := false
 	b.remotes.Range(func(key, value interface{}) bool {
 		v, ok := value.(*client)
 		if ok {
 			if v.route.remoteUrl == url {
 				v.route.remoteID = remoteID
-				exist = true
+				exists = true
 				return false
 			}
 		}
 		return true
 	})
-	return exist
+	return exists
 }
 
 func (b *Broker) SendLocalSubsToRouter(c *client) {
@@ -575,32 +611,28 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 		return true
 	})
 	if len(subInfo.Topics) > 0 {
-		err := c.WriterPacket(subInfo)
+		if err := c.WriterPacket(subInfo); err != nil {
-		if err != nil {
+			log.Error("Send localsubs To Router error", zap.Error(err))
-			log.Error("Send localsubs To Router error :", zap.Error(err))
 		}
 	}
 }
 
 func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacket) {
 	b.routes.Range(func(key, value interface{}) bool {
-		r, ok := value.(*client)
+		if r, ok := value.(*client); ok {
-		if ok {
 			if r.route.remoteID == remoteID {
 				return true
 			}
 			r.WriterPacket(msg)
 		}
 		return true
-
 	})
 }
 
 func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
 
 	b.routes.Range(func(key, value interface{}) bool {
-		r, ok := value.(*client)
+		if r, ok := value.(*client); ok {
-		if ok {
 			r.WriterPacket(packet)
 		}
 		return true
@@ -627,21 +659,32 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 	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))
+		log.Error("search sub client error", zap.Error(err))
 		return
 	}
 
 	for _, sub := range subs {
 		s, ok := sub.(*subscription)
 		if ok {
-			err := s.client.WriterPacket(packet)
+			if err := s.client.WriterPacket(packet); err != nil {
-			if err != nil {
+				log.Error("write message error", zap.Error(err))
-				log.Error("write message error,  ", zap.Error(err))
 			}
 		}
 	}
 }
 
+func (b *Broker) PublishMessageByClientId(packet *packets.PublishPacket, clientId string) error {
+	cli, loaded := b.clients.LoadAndDelete(clientId)
+	if !loaded {
+		return fmt.Errorf("clientId %s not connected", clientId)
+	}
+	conn, success := cli.(*client)
+	if !success {
+		return fmt.Errorf("clientId %s loaded fail", clientId)
+	}
+	return conn.WriterPacket(packet)
+}
+
 func (b *Broker) BroadcastUnSubscribe(topicsToUnSubscribeFrom []string) {
 	if len(topicsToUnSubscribeFrom) == 0 {
 		return

broker/client.go

@@ -248,7 +248,7 @@ func validatePacketFields(msgPacket packets.ControlPacket) (validFields bool) {
 		}
 	}
 
-	// All fields has been validated successfully
+	// All fields have been validated successfully
 	validFields = true
 
 	return
@@ -413,8 +413,8 @@ func (c *client) processClientPublish(packet *packets.PublishPacket) {
 		return
 	}
 
-	//publish kafka
+	//publish to bridge mq
-	c.broker.Publish(&bridge.Elements{
+	cost := c.broker.Publish(&bridge.Elements{
 		ClientID:  c.info.clientID,
 		Username:  c.info.username,
 		Action:    bridge.Publish,
@@ -423,6 +423,10 @@ func (c *client) processClientPublish(packet *packets.PublishPacket) {
 		Topic:     topic,
 	})
 
+	if cost {
+		return
+	}
+
 	switch packet.Qos {
 	case QosAtMostOnce:
 		c.ProcessPublishMessage(packet)
@@ -474,7 +478,6 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 		return
 	}
 
-	// fmt.Println("psubs num: ", len(c.subs))
 	if len(c.subs) == 0 {
 		return
 	}
@@ -808,9 +811,10 @@ func (c *client) Close() {
 		Timestamp: time.Now().Unix(),
 	})
 
-	if c.conn != nil {
+	if c.mu.Lock(); c.conn != nil {
 		_ = c.conn.Close()
 		c.conn = nil
+		c.mu.Unlock()
 	}
 
 	if b == nil {

broker/comm.go

@@ -1,11 +1,10 @@
 package broker
 
 import (
-	"encoding/json"
 	"reflect"
 	"time"
 
-	"github.com/tidwall/gjson"
+	jsoniter "github.com/json-iterator/go"
 	"go.uber.org/zap"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
@@ -134,8 +133,8 @@ func wrapPublishPacket(packet *packets.PublishPacket) *packets.PublishPacket {
 
 func unWrapPublishPacket(packet *packets.PublishPacket) *packets.PublishPacket {
 	p := packet.Copy()
-	if gjson.GetBytes(p.Payload, "payload").Exists() {
+	if payload := jsoniter.Get(p.Payload, "payload").ToString(); payload != "" {
-		p.Payload = []byte(gjson.GetBytes(p.Payload, "payload").String())
+		p.Payload = []byte(payload)
 	}
 	return p
 }
@@ -164,9 +163,14 @@ func publish(sub *subscription, packet *packets.PublishPacket) {
 
 // timer for retry delivery
 func (c *client) ensureRetryTimer(interval ...int64) {
+
+	c.retryTimerLock.Lock()
+	defer c.retryTimerLock.Unlock()
+
 	if c.retryTimer != nil {
 		return
 	}
+
 	if len(interval) > 1 {
 		return
 	}
@@ -174,29 +178,33 @@ func (c *client) ensureRetryTimer(interval ...int64) {
 	if len(interval) == 1 {
 		timerInterval = interval[0]
 	}
-	c.retryTimerLock.Lock()
+
 	c.retryTimer = time.AfterFunc(time.Duration(timerInterval)*time.Second, c.retryDelivery)
-	c.retryTimerLock.Unlock()
+
 	return
 }
 
 func (c *client) resetRetryTimer() {
+	// lock mutex before reading retryTimer
+	c.retryTimerLock.Lock()
+	defer c.retryTimerLock.Unlock()
+
 	if c.retryTimer == nil {
 		return
 	}
-	// reset timer
-	c.retryTimerLock.Lock()
-	c.retryTimer = nil
-	c.retryTimerLock.Unlock()
 
+	// reset timer
+	c.retryTimer = nil
 }
 
 func (c *client) retryDelivery() {
 	c.resetRetryTimer()
 	c.inflightMu.RLock()
 	ilen := len(c.inflight)
-	if c.conn == nil || ilen == 0 { //Reset timer when client offline OR inflight is empty
+
+	if c.mu.Lock(); c.conn == nil || ilen == 0 { //Reset timer when client offline OR inflight is empty
 		c.inflightMu.RUnlock()
+		c.mu.Unlock()
 		return
 	}
 

broker/config.go

@@ -3,7 +3,6 @@ package broker
 import (
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -13,9 +12,12 @@ import (
 	"github.com/fhmq/hmq/logger"
 	"github.com/fhmq/hmq/plugins/auth"
 	"github.com/fhmq/hmq/plugins/bridge"
+	jsoniter "github.com/json-iterator/go"
 	"go.uber.org/zap"
 )
 
+var json = jsoniter.ConfigCompatibleWithStandardLibrary
+
 type Config struct {
 	Worker   int       `json:"workerNum"`
 	HTTPPort string    `json:"httpPort"`

broker/info.go

@@ -15,7 +15,7 @@ func (c *client) SendInfo() {
 	}
 	url := c.info.localIP + ":" + c.broker.config.Cluster.Port
 
-	infoMsg := NewInfo(c.broker.id, url, false)
+	infoMsg := NewInfo(c.broker.id, url)
 	err := c.WriterPacket(infoMsg)
 	if err != nil {
 		log.Error("send info message error, ", zap.Error(err))
@@ -60,7 +60,7 @@ func (c *client) SendConnect() {
 	log.Info("send connect success")
 }
 
-func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
+func NewInfo(sid, url string) *packets.PublishPacket {
 	pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
 	pub.Qos = 0
 	pub.TopicName = BrokerInfoTopic

go.mod

@@ -5,14 +5,14 @@ go 1.18
 require (
 	github.com/Shopify/sarama v1.38.1
 	github.com/bitly/go-simplejson v0.5.0
+	github.com/cespare/xxhash/v2 v2.1.2
 	github.com/eapache/queue v1.1.0
 	github.com/eclipse/paho.mqtt.golang v1.4.2
 	github.com/gin-gonic/gin v1.8.2
 	github.com/google/uuid v1.3.0
+	github.com/json-iterator/go v1.1.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/segmentio/fasthash v1.0.3
 	github.com/stretchr/testify v1.8.1
-	github.com/tidwall/gjson v1.14.4
 	go.uber.org/zap v1.24.0
 	golang.org/x/net v0.7.0
 )
@@ -36,7 +36,6 @@ require (
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.3 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.15.15 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
@@ -47,8 +46,6 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
-	github.com/tidwall/match v1.1.1 // indirect
-	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/ugorji/go/codec v1.2.9 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect

go.sum

@@ -6,6 +6,8 @@ github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkN
 github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -91,8 +93,6 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
-github.com/segmentio/fasthash v1.0.3 h1:EI9+KE1EwvMLBWwjpRDc+fEM+prwxDYbslddQGtrmhM=
-github.com/segmentio/fasthash v1.0.3/go.mod h1:waKX8l2N8yckOgmSsXJi7x1ZfdKZ4x7KRMzBtS3oedY=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -103,13 +103,6 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/tidwall/gjson v1.14.4 h1:uo0p8EbA09J7RQaflQ1aBRffTR7xedD2bcIVSYxLnkM=
-github.com/tidwall/gjson v1.14.4/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
-github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
-github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
-github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
-github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
-github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/ugorji/go/codec v1.2.9 h1:rmenucSohSTiyL09Y+l2OCk+FrMxGMzho2+tjr5ticU=
 github.com/ugorji/go/codec v1.2.9/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
 go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=

main.go

@@ -1,29 +1,30 @@
 package main
 
 import (
-	"log"
 	"os"
 	"os/signal"
-	"runtime"
 
 	"github.com/fhmq/hmq/broker"
+	"github.com/fhmq/hmq/logger"
+	"go.uber.org/zap"
 )
 
+var log = logger.Get()
+
 func main() {
-	runtime.GOMAXPROCS(runtime.NumCPU())
 	config, err := broker.ConfigureConfig(os.Args[1:])
 	if err != nil {
-		log.Fatal("configure broker config error: ", err)
+		log.Fatal("configure broker config error", zap.Error(err))
 	}
 
 	b, err := broker.NewBroker(config)
 	if err != nil {
-		log.Fatal("New Broker error: ", err)
+		log.Fatal("New Broker error: ", zap.Error(err))
 	}
 	b.Start()
 
 	s := waitForSignal()
-	log.Println("signal received, broker closed.", s)
+	log.Info("signal received, broker closed.", zap.Any("signal", s))
 }
 
 func waitForSignal() os.Signal {

plugins/bridge/bridge.go

@@ -37,7 +37,8 @@ const (
 )
 
 type BridgeMQ interface {
-	Publish(e *Elements) error
+	// Publish return true to cost the message
+	Publish(e *Elements) (bool, error)
 }
 
 func NewBridgeMQ(name string) BridgeMQ {

plugins/bridge/csvlog.go

@@ -129,10 +129,9 @@ func (c *csvLog) writeToLog(els []Elements) error {
 // for performance we batch messages into an outqueue and write them in bulk when a timer expires
 func (c *csvLog) Worker() {
 	log.Info("Running CSVLog worker")
-	run := true
 	var outqueue []Elements
 
-	for run == true {
+	for true {
 		c.RLock()
 		waitInterval := c.config.WriteIntervalSecs
 		c.RUnlock()
@@ -191,10 +190,7 @@ func (c *csvLog) Worker() {
 			}
 			break
 		}
-		if run != true {
+
-			log.Info("Closing CSV Bridge worker")
-			break
-		}
 	}
 }
 
@@ -357,7 +353,7 @@ func (c *csvLog) logFilePrune() error {
 // Publish implements the bridge interface - it accepts an Element then checks to see if that element is a
 // message published to the admin topic for the plugin
 //
-func (c *csvLog) Publish(e *Elements) error {
+func (c *csvLog) Publish(e *Elements) (bool, error) {
 	// A short-lived lock on c allows us to
 	// get the Command topic then release the lock
 	// This then allows us to process the command - which may
@@ -376,7 +372,7 @@ func (c *csvLog) Publish(e *Elements) error {
 	// If the outfile is set to "{NULL}" we don't do anything with the message - we just return nil
 	// This feature is here to allow CSVLOG to be enabled/disabled at runtime
 	if OutFile == "{NULL}" {
-		return nil
+		return false, nil
 	}
 
 	if e.Topic == CommandTopic {
@@ -414,5 +410,5 @@ func (c *csvLog) Publish(e *Elements) error {
 	// Push the message into the channel and return
 	// the channel is buffered and is read by a goroutine so this should block for the shortest possible time
 	c.msgchan <- e
-	return nil
+	return false, nil
 }

plugins/bridge/kafka.go

@@ -63,7 +63,7 @@ func (k *kafka) connect() {
 }
 
 //Publish publish to kafka
-func (k *kafka) Publish(e *Elements) error {
+func (k *kafka) Publish(e *Elements) (bool, error) {
 	config := k.kafkaConfig
 	key := e.ClientID
 	topics := make(map[string]bool)
@@ -96,10 +96,10 @@ func (k *kafka) Publish(e *Elements) error {
 			topics[config.DisconnectTopic] = true
 		}
 	default:
-		return errors.New("error action: " + e.Action)
+		return false, errors.New("error action: " + e.Action)
 	}
 
-	return k.publish(topics, key, e)
+	return false, k.publish(topics, key, e)
 
 }
 

plugins/bridge/mock.go

@@ -2,6 +2,6 @@ package bridge
 
 type mockMQ struct{}
 
-func (m *mockMQ) Publish(e *Elements) error {
+func (m *mockMQ) Publish(e *Elements) (bool, error) {
-	return nil
+	return false, nil
 }

pool/fixpool.go

@@ -1,7 +1,7 @@
 package pool
 
 import (
-	"github.com/segmentio/fasthash/fnv1a"
+	"github.com/cespare/xxhash/v2"
 )
 
 type WorkerPool struct {
@@ -29,7 +29,7 @@ func New(maxWorkers int) *WorkerPool {
 }
 
 func (p *WorkerPool) Submit(uid string, task func()) {
-	idx := fnv1a.HashString64(uid) % uint64(p.maxWorkers)
+	idx := xxhash.Sum64([]byte(uid)) % uint64(p.maxWorkers)
 	if task != nil {
 		p.taskQueue[idx] <- task
 	}