collab/hmq
3
0 Fork 0
mirror of https://github.com/fhmq/hmq.git synced 2026-11-05 22:59:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix nil pointer exception in addr logs (#157)

Browse Source
This commit is contained in:
muXxer authored and GitHub committed 2022-04-09 12:53:43 +08:00
1 parent fa7bf33c60
commit c53d8f8a0d
1 file changed
+76 -59
Download Patch File Download Diff File Expand all files Collapse all files
broker/broker.go
+76 -59
View File
@@ -16,6 +16,7 @@ import (
"github.com/eclipse/paho.mqtt.golang/packets" "github.com/eclipse/paho.mqtt.golang/packets"
"go.uber.org/zap" "go.uber.org/zap"
"go.uber.org/zap/zapcore"
"golang.org/x/net/websocket" "golang.org/x/net/websocket"
) )
@@ -55,6 +56,24 @@ func newMessagePool() []chan *Message {
return pool return pool
} }
func getAdditionalLogFields(clientIdentifier string, conn net.Conn, additionalFields ...zapcore.Field) []zapcore.Field {
result := []zapcore.Field{}
// 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 conn != nil && conn.RemoteAddr() != nil {
result = append(result, zap.Stringer("addr", conn.RemoteAddr()))
}
return result
}
func NewBroker(config *Config) (*Broker, error) { func NewBroker(config *Config) (*Broker, error) {
if config == nil { if config == nil {
config = DefaultConfig config = DefaultConfig
@@ -163,7 +182,7 @@ func (b *Broker) StartWebsocketListening() {
err = http.ListenAndServe(hp, mux) err = http.ListenAndServe(hp, mux)
} }
if err != nil { if err != nil {
log.Error("ListenAndServe:" + err.Error()) log.Error("ListenAndServe" + err.Error())
return return
} }
} }
@@ -190,33 +209,39 @@ func (b *Broker) StartClientListening(Tls bool) {
l, err = net.Listen("tcp", hp) l, err = net.Listen("tcp", hp)
log.Info("Start Listening client on ", zap.String("hp", 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 break // successfully listening
} }
log.Error("Error listening on ", zap.Error(err))
time.Sleep(1 * time.Second)
} }
tmpDelay := 10 * ACCEPT_MIN_SLEEP tmpDelay := 10 * ACCEPT_MIN_SLEEP
for { for {
conn, err := l.Accept() conn, err := l.Accept()
if err != nil { if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() { 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) time.Sleep(tmpDelay)
tmpDelay *= 2 tmpDelay *= 2
if tmpDelay > ACCEPT_MAX_SLEEP { if tmpDelay > ACCEPT_MAX_SLEEP {
tmpDelay = ACCEPT_MAX_SLEEP tmpDelay = ACCEPT_MAX_SLEEP
} }
} else { } else {
log.Error("Accept error: %v", zap.Error(err)) log.Error("Accept error", zap.Error(err))
} }
continue continue
} }
tmpDelay = ACCEPT_MIN_SLEEP tmpDelay = ACCEPT_MIN_SLEEP
go b.handleConnection(CLIENT, conn) go b.handleConnection(CLIENT, conn)
} }
} }
@@ -226,7 +251,7 @@ func (b *Broker) StartClusterListening() {
l, e := net.Listen("tcp", hp) l, e := net.Listen("tcp", hp)
if e != nil { if e != nil {
log.Error("Error listening on ", zap.Error(e)) log.Error("Error listening on", zap.Error(e))
return return
} }
@@ -235,15 +260,19 @@ func (b *Broker) StartClusterListening() {
conn, err := l.Accept() conn, err := l.Accept()
if err != nil { if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() { 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) time.Sleep(tmpDelay)
tmpDelay *= 2 tmpDelay *= 2
if tmpDelay > ACCEPT_MAX_SLEEP { if tmpDelay > ACCEPT_MAX_SLEEP {
tmpDelay = ACCEPT_MAX_SLEEP tmpDelay = ACCEPT_MAX_SLEEP
} }
} else { } else {
log.Error("Accept error: %v", zap.Error(err)) log.Error("Accept error", zap.Error(err))
} }
continue continue
} }
@@ -269,7 +298,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
//process connect packet //process connect packet
packet, err := packets.ReadPacket(conn) packet, err := packets.ReadPacket(conn)
if err != nil { if err != nil {
log.Error("read connect packet error: ", zap.Error(err)) log.Error("read connect packet error", zap.Error(err))
conn.Close() conn.Close()
return return
} }
@@ -283,7 +312,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
return return
} }
log.Info("read connect from ", zap.String("clientID", msg.ClientIdentifier), zap.String("addr", conn.RemoteAddr().String())) log.Info("read connect from ", getAdditionalLogFields(msg.ClientIdentifier, conn)...)
connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket) connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
connack.SessionPresent = msg.CleanSession connack.SessionPresent = msg.CleanSession
@@ -292,9 +321,8 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
if connack.ReturnCode != packets.Accepted { if connack.ReturnCode != packets.Accepted {
func() { func() {
defer conn.Close() 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 return
@@ -304,17 +332,15 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
connack.ReturnCode = packets.ErrRefusedNotAuthorised connack.ReturnCode = packets.ErrRefusedNotAuthorised
func() { func() {
defer conn.Close() 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 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 return
} }
@@ -345,25 +371,22 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
c.init() 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 return
} }
cid := c.info.clientID cid := c.info.clientID
var exist bool var exists bool
var old interface{} var old interface{}
switch typ { switch typ {
case CLIENT: 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 {
zap.String("addr", conn.RemoteAddr().String())) log.Warn("client exists, close old client", getAdditionalLogFields(ol.info.clientID, ol.conn)...)
ol, ok := old.(*client)
if ok {
ol.Close() ol.Close()
} }
} }
@@ -379,11 +402,10 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
}) })
} }
case ROUTER: 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() ol.Close()
} }
} }
@@ -400,8 +422,8 @@ func (b *Broker) ConnectToDiscovery() {
for { for {
conn, err = net.Dial("tcp", b.config.Router) conn, err = net.Dial("tcp", b.config.Router)
if err != nil { 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...") log.Debug("Connect to route timeout, retry...")
if 0 == tempDelay { if 0 == tempDelay {
tempDelay = 1 * time.Second tempDelay = 1 * time.Second
@@ -417,7 +439,7 @@ func (b *Broker) ConnectToDiscovery() {
} }
break 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 cid := b.id
info := info{ info := info{
@@ -467,13 +489,13 @@ func (b *Broker) connectRouter(id, addr string) {
conn, err = net.Dial("tcp", addr) conn, err = net.Dial("tcp", addr)
if err != nil { 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 { if retryTimes > 50 {
return return
} }
log.Debug("Connect to route timeout ,retry...") log.Debug("Connect to route timeout, retry...")
if 0 == timeDelay { if 0 == timeDelay {
timeDelay = 1 * time.Second timeDelay = 1 * time.Second
@@ -541,19 +563,19 @@ func (b *Broker) checkNodeExist(id, url string) bool {
} }
func (b *Broker) CheckRemoteExist(remoteID, url string) bool { func (b *Broker) CheckRemoteExist(remoteID, url string) bool {
exist := false exists := false
b.remotes.Range(func(key, value interface{}) bool { b.remotes.Range(func(key, value interface{}) bool {
v, ok := value.(*client) v, ok := value.(*client)
if ok { if ok {
if v.route.remoteUrl == url { if v.route.remoteUrl == url {
v.route.remoteID = remoteID v.route.remoteID = remoteID
exist = true exists = true
return false return false
} }
} }
return true return true
}) })
return exist return exists
} }
func (b *Broker) SendLocalSubsToRouter(c *client) { func (b *Broker) SendLocalSubsToRouter(c *client) {
@@ -576,32 +598,28 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
return true return true
}) })
if len(subInfo.Topics) > 0 { 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) { func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacket) {
b.routes.Range(func(key, value interface{}) bool { 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 { if r.route.remoteID == remoteID {
return true return true
} }
r.WriterPacket(msg) r.WriterPacket(msg)
} }
return true return true
}) })
} }
func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) { func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
b.routes.Range(func(key, value interface{}) bool { b.routes.Range(func(key, value interface{}) bool {
r, ok := value.(*client) if r, ok := value.(*client); ok {
if ok {
r.WriterPacket(packet) r.WriterPacket(packet)
} }
return true return true
@@ -628,16 +646,15 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
err := b.topicsMgr.Subscribers([]byte(packet.TopicName), packet.Qos, &subs, &qoss) err := b.topicsMgr.Subscribers([]byte(packet.TopicName), packet.Qos, &subs, &qoss)
b.mu.Unlock() b.mu.Unlock()
if err != nil { if err != nil {
log.Error("search sub client error, ", zap.Error(err)) log.Error("search sub client error", zap.Error(err))
return return
} }
for _, sub := range subs { for _, sub := range subs {
s, ok := sub.(*subscription) s, ok := sub.(*subscription)
if ok { 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))
} }
} }
} }