147 lines
7.3 KiB
Go
147 lines
7.3 KiB
Go
package wsgateway // 包名声明：与 envelope.go、connection.go、server.go 同包
|
||
|
||
import (
|
||
"context" // 标准库：handler 的入参上下文，连接关闭即取消
|
||
"log/slog" // 标准库：结构化日志，未注册消息类型时打 debug 日志
|
||
"sync" // 标准库：RWMutex，保护 Hub 内部的表与回调列表
|
||
|
||
"git.lzcwy.cn/zhengchiliu/zt-online-game-backend/global" // 本项目：读取 global.WSSetting 判断是否允许多设备
|
||
)
|
||
|
||
// Handler 是业务消息处理器签名，由 hall / room 模块实现并注册到 Hub。
|
||
// ctx 随连接关闭而取消：handler 内部若起 goroutine，必须监听它；绝不要把 *gin.Context 带进来。
|
||
type Handler func(ctx context.Context, c *Connection, env *Envelope)
|
||
|
||
// DefaultHub 是进程级默认 Hub 单例。
|
||
// 若将来需要注入（方便单测），可改为在 main 里 NewHub() 并传给 routers.NewRouter。
|
||
var DefaultHub = NewHub() // 包初始化时即创建，server.go 与路由共用这一个实例
|
||
|
||
// Hub 是连接注册表与消息路由器，本身不理解任何玩法。
|
||
type Hub struct {
|
||
mu sync.RWMutex // 读写锁：保护下面三张表/列表
|
||
conns map[int64]*Connection // userID -> 连接：一个用户当前对应的那条连接
|
||
handlers map[string]Handler // 消息类型 -> 处理器：决定上行消息由谁处理
|
||
disconnectHooks []func(userID int64) // 断连回调：连接移除后通知业务层（房间据此进入重连宽限）
|
||
}
|
||
|
||
// NewHub 构造一个空的 Hub。
|
||
func NewHub() *Hub {
|
||
return &Hub{
|
||
conns: make(map[int64]*Connection), // 初始化连接表，避免 nil map 写入 panic
|
||
handlers: make(map[string]Handler), // 初始化处理器表，避免 nil map 写入 panic
|
||
}
|
||
}
|
||
|
||
// RegisterHandler 注册某类消息的处理函数。
|
||
// 已由 hall 注册 hall.match / hall.cancel；由 room 注册 room.join / room.action / room.leave 等。
|
||
func (h *Hub) RegisterHandler(msgType string, fn Handler) {
|
||
h.mu.Lock() // 加写锁：修改 handlers 表
|
||
defer h.mu.Unlock() // 函数返回时解锁
|
||
h.handlers[msgType] = fn // 覆盖式注册：同名类型以最后注册者为准
|
||
}
|
||
|
||
// OnDisconnect 注册断连回调：连接从表中移除后触发（业务层据此处理掉线）。
|
||
func (h *Hub) OnDisconnect(fn func(userID int64)) {
|
||
h.mu.Lock() // 加写锁：修改回调列表
|
||
defer h.mu.Unlock() // 返回时解锁
|
||
h.disconnectHooks = append(h.disconnectHooks, fn) // 追加回调
|
||
}
|
||
|
||
// Register 把新建连接登记到 Hub；MultiDevice=false 时先踢掉同账号的旧连接。
|
||
func (h *Hub) Register(c *Connection) {
|
||
h.mu.Lock() // 加写锁：可能修改 conns 表
|
||
defer h.mu.Unlock() // 返回时解锁
|
||
|
||
if global.WSSetting == nil || !global.WSSetting.MultiDevice { // 未配置或明确禁止多设备：执行踢线
|
||
if old, ok := h.conns[c.UserID()]; ok && old != c { // 该账号已在线且不是同一条连接
|
||
slog.InfoContext(c.Context(), "ws kick old connection", "user_id", c.UserID()) // 记录踢线日志
|
||
old.Close() // 关闭旧连接：旧连接的读写泵会随之退出
|
||
delete(h.conns, c.UserID()) // 从表中移除旧连接，给新连接腾位置
|
||
}
|
||
}
|
||
h.conns[c.UserID()] = c // 登记新连接
|
||
}
|
||
|
||
// Unregister 从表中注销连接；只有表中仍是自己时才删除，避免误删踢线后新建的连接。
|
||
// 回调在解锁之后执行：避免业务回调内部再次操作 Hub 造成死锁。
|
||
func (h *Hub) Unregister(c *Connection) {
|
||
h.mu.Lock() // 加写锁：修改 conns 表
|
||
var hooks []func(int64) // 待触发的回调
|
||
if cur, ok := h.conns[c.UserID()]; ok && cur == c { // 表中存在且指向同一条连接
|
||
delete(h.conns, c.UserID()) // 安全删除
|
||
hooks = append(hooks, h.disconnectHooks...) // 拷贝回调列表
|
||
}
|
||
h.mu.Unlock() // 解锁后再执行回调
|
||
for _, fn := range hooks { // 逐个触发断连回调
|
||
fn(c.UserID()) // 通知业务层（如房间标记掉线）
|
||
}
|
||
}
|
||
|
||
// Get 取指定用户当前在线的连接。
|
||
func (h *Hub) Get(userID int64) (*Connection, bool) {
|
||
h.mu.RLock() // 加读锁：只读不写
|
||
defer h.mu.RUnlock() // 返回时解锁
|
||
c, ok := h.conns[userID] // 查表
|
||
return c, ok // 返回连接与是否在线
|
||
}
|
||
|
||
// Online 返回当前在线连接数（默认一人一条连接，即在线用户数）。
|
||
func (h *Hub) Online() int {
|
||
h.mu.RLock() // 加读锁
|
||
defer h.mu.RUnlock() // 返回时解锁
|
||
return len(h.conns) // 表长度即在线数
|
||
}
|
||
|
||
// Route 按消息类型分发上行消息；未注册的类型直接回错误，不静默丢弃，方便客户端自测。
|
||
func (h *Hub) Route(ctx context.Context, c *Connection, env *Envelope) {
|
||
h.mu.RLock() // 加读锁：查 handlers 表
|
||
fn, ok := h.handlers[env.Type] // 按类型找处理器
|
||
h.mu.RUnlock() // 立即解锁，避免业务执行期间占用锁
|
||
|
||
if !ok { // 没找到对应处理器
|
||
slog.DebugContext(ctx, "ws unsupported message type", "type", env.Type, "user_id", c.UserID()) // 记录调试日志
|
||
c.SendError(env.Seq, 0, "暂不支持的消息类型: "+env.Type) // 回一条错误信封，带上原 seq 便于客户端定位
|
||
return // 结束分发
|
||
}
|
||
fn(ctx, c, env) // 交给业务处理器执行
|
||
}
|
||
|
||
// SendTo 向单个用户下发一条已编码的消息；返回 false 表示已离线或发送队列已满。
|
||
func (h *Hub) SendTo(userID int64, data []byte) bool {
|
||
h.mu.RLock() // 加读锁：查表
|
||
c, ok := h.conns[userID] // 找到目标连接
|
||
h.mu.RUnlock() // 解锁
|
||
if !ok { // 用户不在线
|
||
return false // 返回投递失败，调用方可决定是否落库或重连后再发
|
||
}
|
||
return c.Send(data) // 入队发送
|
||
}
|
||
|
||
// Broadcast 向指定用户集合下发同一条消息（房间内广播与私有事件都走这里）。
|
||
func (h *Hub) Broadcast(userIDs []int64, data []byte) {
|
||
h.mu.RLock() // 加读锁
|
||
targets := make([]*Connection, 0, len(userIDs)) // 预分配切片，容量等于目标数
|
||
for _, uid := range userIDs { // 遍历目标用户 ID
|
||
if c, ok := h.conns[uid]; ok { // 在线才加入投递列表
|
||
targets = append(targets, c) // 收集连接指针
|
||
}
|
||
}
|
||
h.mu.RUnlock() // 解锁：后续发送不持锁，避免慢客户端阻塞注册表
|
||
for _, c := range targets { // 逐个投递
|
||
c.Send(data) // 入队发送（队列满会自行断开慢客户端）
|
||
}
|
||
}
|
||
|
||
// BroadcastAll 向全部在线连接下发，适用于大厅公告、全服通知等。
|
||
func (h *Hub) BroadcastAll(data []byte) {
|
||
h.mu.RLock() // 加读锁
|
||
targets := make([]*Connection, 0, len(h.conns)) // 预分配切片，容量等于当前在线数
|
||
for _, c := range h.conns { // 遍历所有连接
|
||
targets = append(targets, c) // 收集连接指针
|
||
}
|
||
h.mu.RUnlock() // 解锁后再发送
|
||
for _, c := range targets { // 逐个投递
|
||
c.Send(data) // 入队发送
|
||
}
|
||
}
|