修复和优化日志字段的获取逻辑,确保在不同连接类型下正确获取远程地址。同时,调整注释格式以提高代码可读性。

This commit is contained in:
joy,zhou
2025-07-28 09:09:16 +08:00
parent c6675ca143
commit c5b4c362a2
+21 -21
View File
@@ -81,10 +81,16 @@ func getAdditionalLogFields(clientIdentifier string, conn net.Conn, additionalFi
result = append(result, zap.String("clientID", clientIdentifier)) result = append(result, zap.String("clientID", clientIdentifier))
// add remote connection address // add remote connection address
if !wsEnabled && conn != nil && conn.RemoteAddr() != nil { if !wsEnabled && conn != nil {
result = append(result, zap.Stringer("addr", conn.RemoteAddr())) if conn.RemoteAddr() != nil {
} else if wsEnabled && wsConn != nil && wsConn.Request() != nil { result = append(result, zap.String("addr", conn.RemoteAddr().String()))
result = append(result, zap.String("addr", wsConn.Request().RemoteAddr)) }
} else if wsEnabled && wsConn != nil {
if wsConn.Request() != nil {
result = append(result, zap.String("addr", wsConn.Request().RemoteAddr))
} else {
result = append(result, zap.String("addr", wsConn.RemoteAddr().String()))
}
} }
return result return result
@@ -143,7 +149,6 @@ func (b *Broker) SubmitWork(clientId string, msg *Message) {
ProcessMessage(msg) ProcessMessage(msg)
}) })
} }
} }
func (b *Broker) Start() { func (b *Broker) Start() {
@@ -156,41 +161,40 @@ func (b *Broker) Start() {
go InitHTTPMoniter(b) go InitHTTPMoniter(b)
} }
//listen client over tcp // listen client over tcp
if b.config.Port != "" { if b.config.Port != "" {
go b.StartClientListening(false) go b.StartClientListening(false)
} }
//listen client over unix // listen client over unix
if b.config.Port == "" && b.config.UnixFilePath != "" { if b.config.Port == "" && b.config.UnixFilePath != "" {
go b.StartUnixSocketClientListening(b.config.UnixFilePath, true) go b.StartUnixSocketClientListening(b.config.UnixFilePath, true)
} }
//listen client over windows pipe // listen client over windows pipe
if b.config.Port == "" && b.config.UnixFilePath == "" && b.config.WindowsPipeName != "" { if b.config.Port == "" && b.config.UnixFilePath == "" && b.config.WindowsPipeName != "" {
go b.StartPipeSocketListening(b.config.WindowsPipeName, true) go b.StartPipeSocketListening(b.config.WindowsPipeName, true)
} }
//listen for cluster // listen for cluster
if b.config.Cluster.Port != "" { if b.config.Cluster.Port != "" {
go b.StartClusterListening() go b.StartClusterListening()
} }
//listen for websocket // listen for websocket
if b.config.WsPort != "" { if b.config.WsPort != "" {
go b.StartWebsocketListening() go b.StartWebsocketListening()
} }
//listen client over tls // listen client over tls
if b.config.TlsPort != "" { if b.config.TlsPort != "" {
go b.StartClientListening(true) go b.StartClientListening(true)
} }
//connect on other node in cluster // connect on other node in cluster
if b.config.Router != "" { if b.config.Router != "" {
go b.processClusterInfo() go b.processClusterInfo()
b.ConnectToDiscovery() b.ConnectToDiscovery()
} }
} }
func (b *Broker) StartWebsocketListening() { func (b *Broker) StartWebsocketListening() {
@@ -388,7 +392,7 @@ func (b *Broker) DisConnClientByClientId(clientId string) {
} }
func (b *Broker) handleConnection(typ int, conn net.Conn) error { func (b *Broker) handleConnection(typ int, conn net.Conn) error {
//process connect packet // process connect packet
packet, err := packets.ReadPacket(conn) packet, err := packets.ReadPacket(conn)
if err != nil { if err != nil {
return errors.New(fmt.Sprintf("read connect packet error:%v", err)) return errors.New(fmt.Sprintf("read connect packet error:%v", err))
@@ -473,7 +477,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) error {
} }
b.clients.Store(cid, c) b.clients.Store(cid, c)
var pubPack = PubPacket{} pubPack := PubPacket{}
if willmsg != nil { if willmsg != nil {
pubPack.TopicName = info.willMsg.TopicName pubPack.TopicName = info.willMsg.TopicName
pubPack.Payload = info.willMsg.Payload pubPack.Payload = info.willMsg.Payload
@@ -568,7 +572,6 @@ func (b *Broker) processClusterInfo() {
} }
ProcessMessage(msg) ProcessMessage(msg)
} }
} }
func (b *Broker) connectRouter(id, addr string) { func (b *Broker) connectRouter(id, addr string) {
@@ -633,7 +636,6 @@ func (b *Broker) connectRouter(id, addr string) {
go c.readLoop() go c.readLoop()
go c.StartPing() go c.StartPing()
} }
func (b *Broker) checkNodeExist(id, url string) bool { func (b *Broker) checkNodeExist(id, url string) bool {
@@ -646,7 +648,7 @@ func (b *Broker) checkNodeExist(id, url string) bool {
return true return true
} }
//skip // skip
l, ok := v.(string) l, ok := v.(string)
if ok { if ok {
if url == l { if url == l {
@@ -713,7 +715,6 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke
} }
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 {
if r, ok := value.(*client); ok { if r, ok := value.(*client); ok {
r.WriterPacket(packet) r.WriterPacket(packet)
@@ -800,7 +801,7 @@ func (b *Broker) OnlineOfflineNotification(info Info, online bool, lastMsg int64
} }
if b, err := encJson.Marshal(msg); err != nil { if b, err := encJson.Marshal(msg); err != nil {
//This is a TERRIBLE situation, falling back to legacy format to not break API Contract // This is a TERRIBLE situation, falling back to legacy format to not break API Contract
packet.Payload = []byte(fmt.Sprintf(`{"clientID":"%s","online":%v,"timestamp":"%s"}`, info.ClientID, online, time.Now().UTC().Format(time.RFC3339))) packet.Payload = []byte(fmt.Sprintf(`{"clientID":"%s","online":%v,"timestamp":"%s"}`, info.ClientID, online, time.Now().UTC().Format(time.RFC3339)))
} else { } else {
packet.Payload = b packet.Payload = b
@@ -818,5 +819,4 @@ func FileExist(name string) bool {
} else { } else {
panic(err) panic(err)
} }
} }