package wsgateway // 包名声明:服务端连接对象,属于传输层 import ( "context" // 标准库:连接的生命周期上下文,关闭即取消 "errors" // 标准库:用 errors.Is 判断上下文取消、连接关闭等错误 "log/slog" // 标准库:结构化日志,记录异常读写与限流踢线 "net" // 标准库:net.ErrClosed 用于识别「连接已关闭」类错误 "sync" // 标准库:Mutex 保护活跃时间与限流计数,Once 保证关闭幂等 "time" // 标准库:超时时间、心跳周期、限流窗口都依赖时间 "git.lzcwy.cn/zhengchiliu/zt-online-game-backend/global" // 本项目:读取 global.WSSetting 的传输层参数 "github.com/coder/websocket" // 第三方:coder/websocket,负责帧读写、ping、关闭 ) // pongWait 返回「多久没收到消息就判死」的时长,对应配置的 ReadTimeout。 func pongWait() time.Duration { if global.WSSetting != nil && global.WSSetting.ReadTimeout > 0 { // 配置存在且为正数 return time.Duration(global.WSSetting.ReadTimeout) * time.Second // 按配置(秒)换算为 Duration } return 60 * time.Second // 配置缺失时的兜底值,避免 0 值导致连接立刻被判死 } // writeWait 返回单次写或 ping 的超时时长,对应配置的 WriteTimeout。 func writeWait() time.Duration { if global.WSSetting != nil && global.WSSetting.WriteTimeout > 0 { // 配置存在且为正数 return time.Duration(global.WSSetting.WriteTimeout) * time.Second // 按配置换算 } return 10 * time.Second // 兜底 10 秒 } // pingPeriod 返回服务端主动 ping 的周期,对应配置的 PingInterval。 func pingPeriod() time.Duration { if global.WSSetting != nil && global.WSSetting.PingInterval > 0 { // 配置存在且为正数 return time.Duration(global.WSSetting.PingInterval) * time.Second // 按配置换算 } return 25 * time.Second // 兜底 25 秒 } // queueSize 返回发送队列长度,对应配置的 SendQueueSize。 func queueSize() int { if global.WSSetting != nil && global.WSSetting.SendQueueSize > 0 { // 配置存在且为正数 return global.WSSetting.SendQueueSize // 使用配置值 } return 64 // 兜底 64 条 } // Connection 表示服务端持有的一条 WebSocket 连接(不是客户端)。 // 每条连接由两个 goroutine 驱动:ReadPump 收、WritePump 发,中间用 send 队列解耦。 // // 说明:coder/websocket 会自动应答协议级 ping 帧,但该应答对 Read 不可见, // 因此「死连接」判定由 WritePump 侧的空闲检查负责。 type Connection struct { hub *Hub // 所属 Hub:用于注销自己、以及业务层反向查找 conn *websocket.Conn // 底层 ws 连接句柄:真正的帧读写入口 userID int64 // 该连接归属的用户 ID:由建连时的 JWT 解析得到 send chan []byte // 发送队列:业务侧只入队,写 goroutine 负责真正发送 closed chan struct{} // 关闭信号:close 后所有 select 立即感知 once sync.Once // 关闭幂等:Close 可能被读泵、写泵、Hub 多处调用 ctx context.Context // 连接生命周期上下文:关闭时取消,供 handler 起的 goroutine 感知 cancel context.CancelFunc // 取消函数:Close 时调用,触发 ctx 取消 mu sync.Mutex // 保护下面三个字段(活跃时间、限流窗口起点、限流计数) lastActive time.Time // 最近一次收到客户端消息的时间:判死连接用 winStart time.Time // 限流窗口起点:每秒重置一次 winCount int // 当前窗口内已收到的消息数 } // NewConnection 为一条已升级的连接创建对象。 func NewConnection(hub *Hub, conn *websocket.Conn, userID int64) *Connection { ctx, cancel := context.WithCancel(context.Background()) // 以根 context 派生可取消的连接上下文 return &Connection{ hub: hub, // 绑定所属 Hub conn: conn, // 保存底层连接 userID: userID, // 保存用户 ID send: make(chan []byte, queueSize()), // 按配置创建带缓冲的发送队列 closed: make(chan struct{}), // 创建关闭信号通道 ctx: ctx, // 保存上下文 cancel: cancel, // 保存取消函数,供 Close 调用 lastActive: time.Now(), // 初始活跃时间为当前时间,避免刚连上就被判死 } } // UserID 返回该连接对应的用户 ID。 func (c *Connection) UserID() int64 { return c.userID // 直接返回建立连接时保存的 ID } // Context 返回连接生命周期上下文:handler 内起 goroutine 时用它感知断连。 func (c *Connection) Context() context.Context { return c.ctx // 返回内部上下文 } // Send 把一条已编码的消息放入发送队列;队列满视为慢客户端并断开。 func (c *Connection) Send(data []byte) bool { select { case <-c.closed: // 连接已关闭 return false // 投递失败 default: // 未关闭则继续往下走,避免阻塞 } select { case c.send <- data: // 尝试入队成功 return true // 投递成功 case <-c.closed: // 入队期间连接被关闭 return false // 投递失败 default: // 队列已满:说明客户端消费跟不上 slog.WarnContext(c.ctx, "ws send queue full, close slow client", "user_id", c.userID) // 记录慢客户端日志 c.Close() // 主动断开,避免内存被堆积消息撑爆 return false // 投递失败 } } // SendEnvelope 把信封编码后入队发送。 func (c *Connection) SendEnvelope(typ string, seq int64, payload any) bool { data, err := Encode(typ, seq, payload) // 先序列化为 JSON 字节 if err != nil { // 序列化失败(payload 含不可序列化字段) slog.ErrorContext(c.ctx, "ws encode envelope failed", "type", typ, "error", err) // 记录错误日志 return false // 发送失败 } return c.Send(data) // 入队发送 } // SendError 便捷地发一条错误下行消息。 func (c *Connection) SendError(seq int64, code int, msg string) bool { return c.SendEnvelope(TypeError, seq, ErrorPayload{Code: code, Msg: msg}) // 组装 ErrorPayload 并以 error 类型下发 } // Close 关闭连接,多处调用只生效一次。 func (c *Connection) Close() { c.once.Do(func() { // 用 Once 保证幂等,避免重复 close 通道 panic close(c.closed) // 广播关闭信号,读写泵立即退出 c.cancel() // 取消连接上下文,通知所有派生 goroutine _ = c.conn.Close(websocket.StatusNormalClosure, "") // 以正常关闭状态码关闭底层连接,忽略返回错误 }) } // touch 把活跃时间刷新为当前时间,表示刚收到客户端消息。 func (c *Connection) touch() { c.mu.Lock() // 加锁:与 idleFor 共享字段 c.lastActive = time.Now() // 更新活跃时间 c.mu.Unlock() // 解锁 } // idleFor 返回距上次收到消息已经过去多久。 func (c *Connection) idleFor() time.Duration { c.mu.Lock() // 加锁读取 defer c.mu.Unlock() // 函数返回时解锁 return time.Since(c.lastActive) // 计算与当前时间的差值 } // allow 做固定窗口限流:每秒消息数超过 MaxMsgPerSec 就返回 false。 func (c *Connection) allow() bool { limit := 0 // 默认不限流 if global.WSSetting != nil { // 配置存在 limit = global.WSSetting.MaxMsgPerSec // 取配置的每秒上限 } if limit <= 0 { // 未配置或配置为 0 return true // 不做限制 } now := time.Now() // 取当前时间 c.mu.Lock() // 加锁:修改窗口计数 defer c.mu.Unlock() // 返回时解锁 if now.Sub(c.winStart) >= time.Second { // 距窗口起点已超过 1 秒 c.winStart = now // 开启新窗口 c.winCount = 0 // 计数归零 } c.winCount++ // 本条消息计入当前窗口 return c.winCount <= limit // 未超过上限才放行 } // ReadPump 是读循环:收帧 → 更新活跃时间 → 限流 → 解码 → 心跳 → 路由分发。 // 该函数阻塞运行,一旦返回就表示连接结束,defer 里完成注销与关闭。 func (c *Connection) ReadPump() { defer func() { // 退出前必定执行清理 c.hub.Unregister(c) // 从 Hub 表中摘掉自己 c.Close() // 关闭连接(幂等) }() for { // 持续读取,直到出错或连接关闭 mt, data, err := c.conn.Read(c.ctx) // 阻塞读取一帧;coder/websocket 会自动应答协议级 ping,这里只拿到业务消息 if err != nil { // 读取出错:可能是正常关闭、上下文取消或对端异常断开 if !isNormalClose(err) { // 过滤掉正常关闭类错误,避免噪音日志 slog.WarnContext(c.ctx, "ws read error", "user_id", c.userID, "error", err) // 记录异常读错误 } return // 退出读循环 } if mt != websocket.MessageText { // 只接受文本帧(业务信封统一用 JSON 文本) continue // 二进制帧直接忽略,继续读下一条 } c.touch() // 收到消息:刷新活跃时间,避免被判死 if !c.allow() { // 超过每秒消息数上限 slog.WarnContext(c.ctx, "ws message rate exceeded", "user_id", c.userID) // 记录限流日志 c.SendError(0, 0, "消息频率超限") // 先回一条错误告知客户端 return // 再断开连接 } env, err := Decode(data) // 把 JSON 字节解析成统一信封 if err != nil { // 信封格式不合法 c.SendError(0, 0, "消息格式错误: "+err.Error()) // 回错误但不断开,允许客户端继续尝试 continue // 继续读下一条 } if env.Type == TypePing { // 应用层心跳 c.SendEnvelope(TypePong, env.Seq, nil) // 原样带回 seq 回一条 pong continue // 心跳不进业务路由 } // TODO(第二步):大厅与房间的 handler 都从这里进入(hall.match / room.action 等)。 c.hub.Route(c.ctx, c, env) // 按类型分发给已注册的业务处理器 } } // WritePump 是写循环:发送队列 + 主动 ping 保活 + 空闲超时判死。 func (c *Connection) WritePump() { ticker := time.NewTicker(pingPeriod()) // 按配置周期触发心跳与空闲检查 defer func() { // 退出前清理 ticker.Stop() // 停止定时器,避免泄漏 c.Close() // 关闭连接(幂等) }() for { // 循环处理三类事件 select { case <-c.closed: // 连接被关闭 return // 退出写循环 case msg := <-c.send: // 有待发送的下行消息 ctx, cancel := context.WithTimeout(c.ctx, writeWait()) // 为本次写操作加超时,防止对端不读导致永久阻塞 err := c.conn.Write(ctx, websocket.MessageText, msg) // 写出一帧文本消息 cancel() // 立即释放超时上下文资源 if err != nil { // 写失败:连接多半已不可用 return // 退出写循环,由 defer 关闭 } case <-ticker.C: // 心跳周期到点 if idle := c.idleFor(); idle > pongWait() { // 超过 ReadTimeout 未收到任何客户端消息 slog.InfoContext(c.ctx, "ws idle timeout, close connection", "user_id", c.userID, "idle", idle.String()) // 记录判死日志 _ = c.conn.Close(websocket.StatusPolicyViolation, "idle timeout") // 以策略违规码关闭,告知对端原因 return // 退出写循环 } ctx, cancel := context.WithTimeout(c.ctx, writeWait()) // 为 ping 加超时 err := c.conn.Ping(ctx) // 发送协议级 ping,探测链路与对端存活 cancel() // 释放上下文 if err != nil { // ping 失败:连接已死 return // 退出写循环 } } } } // isNormalClose 判断错误是否属于「正常结束」,避免把正常断开记成异常告警。 func isNormalClose(err error) bool { if err == nil { // 没有错误 return true // 视为正常 } if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, net.ErrClosed) { // 上下文被取消(我们主动关闭)、超时、或底层连接已关闭,都属于预期内的结束 return true // 视为正常 } return websocket.CloseStatus(err) != -1 // 有明确关闭状态码说明是对端发起的正常关闭 }