网站搭建免费软件百度广告推广

处理 NATS 订阅的消息

在 WebSocket 的管理模块中，我们之前已经处理了一些消息。这些消息通过 NATS 订阅过来，我们需要对这些消息进行进一步的处理。一旦消息到达，我们需要执行相应的操作，并将结果发送回去，包括之前的操作。

理论上，所有消息都应该能够到达这里进行处理。目前，我们还没有对这部分进行处理，接下来需要完善这部分逻辑。

3. 解析消息并处理

当收到消息后，我们需要解析消息内容。解析完成后，我们需要根据消息的类型（type）进行不同的处理。

• 特殊类型处理：

  • 如果消息类型是“三审”（假设的类型），我们需要进行特殊处理。因为这种类型的消息可能存储在 WebSocket 的连接中，而不是直接推送给客户端。

  • 如果消息类型是“replace”或“response”，我们需要将消息类型改为“response”，以便正确地将消息发送回客户端。

• 普通类型处理：

  • 如果消息类型是“pose”，我们需要将消息放入一个专门的通道（Channel）中进行处理。这样可以提高系统的承载能力，避免消息积压。

4. 消息推送逻辑

在处理完消息后，我们需要将消息推送给客户端。具体步骤如下：

1. 获取客户端 ID：

   • 从消息中获取当前客户端的 ID。如果客户端不存在，我们需要记录日志并提示客户端已下线。

2. 编码消息：

   • 对消息进行编码处理，确保消息格式正确。

3. 发送消息：

   • 将编码后的消息发送给客户端。如果消息类型是“pose”，我们需要循环处理所有相关客户端，并将消息推送给每个客户端。

5. 代码实现

以下是优化后的代码实现：

package websocketimport ("encoding/json""fmt""log""sync""github.com/nats-io/nats.go""github.com/gorilla/websocket"
)type WebSocketManager struct {sync.RWMutexclients    map[*websocket.Conn]stringmessageCh  chan *MessagepushCh     chan *MessagenatsConn   *nats.Conn
}type Message struct {Type    string          `json:"type"`Data    json.RawMessage `json:"data"`ClientID string          `json:"clientID"`
}func NewWebSocketManager(natsConn *nats.Conn) *WebSocketManager {return &WebSocketManager{clients:   make(map[*websocket.Conn]string),messageCh: make(chan *Message, 1024),pushCh:    make(chan *Message, 1024),natsConn:  natsConn,}
}func (wm *WebSocketManager) handleMessage(msg *Message) {log.Printf("Handling message: %+v", msg)// 获取客户端连接clientID := msg.ClientIDclientConn, exists := wm.clients[clientID]if !exists {log.Printf("Client not found: %s", clientID)return}// 根据消息类型处理switch msg.Type {case "response":// 处理响应消息wm.sendToClient(clientConn, msg)case "pose":// 处理推送消息wm.pushCh <- msgdefault:log.Printf("Unknown message type: %s", msg.Type)}
}func (wm *WebSocketManager) sendToClient(clientConn *websocket.Conn, msg *Message) {encodedMsg, err := json.Marshal(msg)if err != nil {log.Printf("Failed to encode message: %v", err)return}err = clientConn.WriteMessage(websocket.TextMessage, encodedMsg)if err != nil {log.Printf("Failed to send message to client: %v", err)}
}func (wm *WebSocketManager) processPushMessages() {for msg := range wm.pushCh {for clientConn := range wm.clients {wm.sendToClient(clientConn, msg)}}
}func (wm *WebSocketManager) start() {go wm.processPushMessages()// NATS 订阅消息nc := wm.natsConnnc.Subscribe("ws.messages", func(msg *nats.Msg) {var wsMsg Messageerr := json.Unmarshal(msg.Data, &wsMsg)if err != nil {log.Printf("Failed to unmarshal NATS message: %v", err)return}wm.handleMessage(&wsMsg)})
}