wssvr

package
v1.1.34 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 18, 2026 License: MIT Imports: 31 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultWsRoute      = "/ws"       // 默认WebSocket路由路径
	DefaultWsMaxBodyLen = 1024 * 1024 // 默认单条消息体最大 1MB，可通过 WsServer.SetMaxBodyLen 覆盖
)

WebSocket服务器实现

错误类型约定：

  • 协议错误（客户端请求不合法、鉴权失败等）：使用 ex.Throw{Code, Msg[, Err]}，便于上层按 HTTP 状态码处理或回写。
  • 配置/内部错误（校验配置、连接池、调用方参数等）：使用 utils.Error，仅作日志或返回给调用方，无状态码需求。

WebSocket专用常量

Variables

This section is empty.

Functions

func PutMessageHandler 

func PutMessageHandler(mh *MessageHandler)

func RemoteIPFromRequest 

func RemoteIPFromRequest(r *http.Request) string

Types

type CacheAware 

type CacheAware func(ds ...string) (cache.Cache, error)

type ConfigValidator 

type ConfigValidator struct{}

ConfigValidator 配置验证器（统一配置检查）

type ConnectionContext 

type ConnectionContext struct {
	Subject      *jwt.Subject
	WsConn       *gws.Conn // WebSocket 连接（gws）
	DevConn      *DevConn
	Server       *WsServer
	RouterConfig *RouterConfig // 路由配置
	Path         string        // WebSocket连接的路径
	ClientIP     string        // 建连时从 HTTP 升级请求解析的真实客户端 IP
	RawToken     []byte        // 原始JWT token字节，用于签名验证
	// contains filtered or unexported fields
}

ConnectionContext 每个 WebSocket 连接的上下文（与 HTTP 的 node.Context 不同，不含 JsonBody）。 单帧协议体使用 Process(..., jb) 的池化 *wire.JsonBody，避免与 ParallelEnabled 下的并发收包共享指针。

func (*ConnectionContext) AssignSessionID 

func (cc *ConnectionContext) AssignSessionID() int64

AssignSessionID 分配新的 WS 会话雪花 ID（Login 成功时调用；同连接重复 Login 会覆盖旧值）。

func (*ConnectionContext) GetRawTokenBytes 

func (cc *ConnectionContext) GetRawTokenBytes() []byte

GetRawTokenBytes 获取原始JWT token字节

func (*ConnectionContext) GetSessionID 

func (cc *ConnectionContext) GetSessionID() int64

GetSessionID 返回 Login 成功后分配的 WS 会话雪花 ID；尚未 Login 时为 0。

func (*ConnectionContext) GetTokenSecret 

func (cc *ConnectionContext) GetTokenSecret() []byte

GetTokenSecret 获取WebSocket连接的 token 派生密钥（每次调用重新派生，不缓存）。 为保证安全需在用毕后 DIC.ClearData(secret)；为此接受每次 HMAC 派生的性能损耗。

func (*ConnectionContext) GetUserID 

func (cc *ConnectionContext) GetUserID() int64

GetUserID 获取用户ID int64类型

func (*ConnectionContext) GetUserIDInt64 

func (cc *ConnectionContext) GetUserIDInt64() int64

GetUserIDInt64 获取用户ID int64类型

func (*ConnectionContext) GetUserIDString 

func (cc *ConnectionContext) GetUserIDString() string

GetUserIDString 获取用户ID string类型

func (*ConnectionContext) Parser 

func (cc *ConnectionContext) Parser(body []byte, dst interface{}) error

Parser 解析 WebSocket 业务数据并填充 dto.BaseReq 上下文。

func (*ConnectionContext) RemoteIP 

func (cc *ConnectionContext) RemoteIP() string

RemoteIP 返回建连时快照的真实客户端 IP（serveHTTP 升级前由 RemoteIPFromRequest 写入 ClientIP）。

type ConnectionManager 

type ConnectionManager struct {
	// contains filtered or unexported fields
}

ConnectionManager 连接管理器：线程安全的连接管理，支持广播、按 subject 推送、过期清理。

设计要点： - conns：二级 map subject -> deviceKey -> *DevConn，deviceKey 由 mode 决定（SubjectUnique 时为 sub，SubjectDeviceUnique 时为 sub_dev）。 - totalConn：原子计数，用于限流、Count()、以及 CleanupExpired/sendToSubjectByJsonResp 的 slice 预分配容量，避免在 RLock 内做重逻辑。 - 所有"收集 conn 再关闭/发送"的路径均在 RLock 内只收集指针，在锁外执行 I/O，避免持锁时间过长。 - reverseIndex：反向索引 *gws.Conn -> reverseIndexEntry，用于 O(1) 复杂度的 RemoveByConn（适合连接数 > 10000 的场景）。

func NewConnectionManager 

func NewConnectionManager(maxConn int, mode ConnectionUniquenessMode, broadcastKeyProvider func(subject string) string) *ConnectionManager

NewConnectionManager 创建连接管理器

func (*ConnectionManager) Add 

func (cm *ConnectionManager) Add(conn *DevConn) error

Add 添加连接 Add 将连接加入管理器。根据连接唯一性模式决定策略： - SubjectUnique: 替换同 subject 的所有连接，只保留一个。 - SubjectDeviceUnique: 替换同 subject+device 的连接，允许多设备同时在线。 设备键格式：subject_device (如: user123_web, user123_app)

设计要点：若存在旧连接，先从 map 移除并减 totalConn，再在 goroutine 中 closeConn，避免锁内 I/O 阻塞。

func (*ConnectionManager) CleanupAll 

func (cm *ConnectionManager) CleanupAll()

CleanupAll 关闭所有连接。先 RLock 内收集全部 conn 指针并预分配 slice，锁外再统一 closeConn。 设计要点：先发送关闭帧，OnClose 回调会负责从 map 中移除（避免重复移除）。

func (*ConnectionManager) CleanupExpired 

func (cm *ConnectionManager) CleanupExpired(timeoutSeconds int64) int

CleanupExpired 清理空闲超过 timeoutSeconds 的连接。

设计要点： - 在 RLock 内仅收集过期 conn 指针（LastSeen 为原子读无锁），RUnlock 后再 closeConn，避免持锁做 I/O。 - toClose 按 totalConn 预分配容量，减少 append 扩容与 GC。 - 超时判断依赖 DevConn.Last（每次收包/心跳 UpdateLast），由 HeartbeatService 按 idleTimeout 周期性调用。 - 先发送关闭帧，OnClose 回调会负责从 map 中移除（避免重复移除）。

func (*ConnectionManager) Count 

func (cm *ConnectionManager) Count() int

Count 获取当前连接数

func (*ConnectionManager) Get 

func (cm *ConnectionManager) Get(subject, deviceKey string) *DevConn

Get 获取指定连接

func (*ConnectionManager) GetAllSubjectDevices 

func (cm *ConnectionManager) GetAllSubjectDevices() map[string][]string

GetAllSubjectDevices 获取所有用户连接subject

func (*ConnectionManager) GetSubjectDevices 

func (cm *ConnectionManager) GetSubjectDevices(subject string) map[string][]string

GetSubjectDevices

func (*ConnectionManager) HealthCheck 

func (cm *ConnectionManager) HealthCheck() map[string]int

HealthCheck 健康检查：返回每个 subject 的连接数。 活性由 LastSeen 与 CleanupExpired 维护。

func (*ConnectionManager) Remove 

func (cm *ConnectionManager) Remove(subject, deviceKey string) *DevConn

Remove 移除连接（不关闭，仅从管理器移除）

func (*ConnectionManager) RemoveByConn 

func (cm *ConnectionManager) RemoveByConn(conn *DevConn) bool

RemoveByConn 按连接指针从管理器中移除该连接。 设计要点： - 使用 reverseIndex 反向索引实现 O(1) 查找，避免遍历所有连接（适合连接数 > 10000 的场景）。 - 必须用指针精确匹配，避免“新连接已替换旧连接”时误把新连接从 map 删掉。 - 关闭由 closeConnFromLoop 等调用方负责。 返回是否成功移除（若连接已被替换则未找到，返回 false）。

func (*ConnectionManager) SendToSubject 

func (cm *ConnectionManager) SendToSubject(subject, router string, data interface{}) error

SendToSubject 按 subject 推送消息（subject=="" 时为全量广播）。

调用约定： - subject: 目标用户标识；传空字符串表示发送给全部在线连接。 - router: 业务路由，必填；客户端通常按该路由分发消息。 - data: 业务载荷，必填；内部会先 JSON 序列化再 Base64 封装到 JsonResp.Data。

安全与协议： - 本方法会统一构造 JsonResp（Code=300, Message="push", Plan=0）。 - 签名密钥来自 broadcastKeyProvider(subject)，支持按用户/广播动态取密钥。 - 未配置可用密钥时返回错误，避免发送未签名或弱签名数据。

type ConnectionUniquenessMode 

type ConnectionUniquenessMode int

ConnectionUniquenessMode 连接唯一性模式 用于控制WebSocket连接的唯一性策略 

const (
	// SubjectUnique 仅Subject唯一，一个用户只能有一个连接
	// 适用于单设备应用场景，如移动端App
	SubjectUnique ConnectionUniquenessMode = iota

	// SubjectDeviceUnique Subject+Device唯一，一个用户可以在多个设备上连接
	// 适用于多设备场景，如Web、App、PC同时在线
	SubjectDeviceUnique
)

type DevConn 

type DevConn struct {
	Sub  string
	Dev  string
	Last int64     // 最后活跃时间戳，原子读写，供 CleanupExpired 无锁判断
	Conn *gws.Conn // WebSocket 连接（gws）
	// contains filtered or unexported fields
}

DevConn 设备连接实体：存储单连接的核心信息。

设计要点： - Last：使用原子读写（UpdateLast 写、LastSeen 读），使 CleanupExpired 遍历时无需加锁。 - 多 goroutine 并发 Send 依赖 gws.Conn.WriteMessage 的线程安全（库内 channel + 单写）；不再额外加 sendMu，避免与 ParallelEnabled 下多回包场景无谓争用。 - closed：与 CleanupExpired / 踢线等路径上的 WriteClose 配合，Send 内先读后写；与 WriteClose 之间无应用层全局锁（历来如此）。 - closeOnce：保证 Close() 只执行一次，避免重复关闭导致 panic。

func (*DevConn) LastSeen 

func (dc *DevConn) LastSeen() int64

LastSeen 返回最近一次活跃时间。原子读、无锁，供 CleanupExpired 在 RLock 内批量调用。

func (*DevConn) Send 

func (dc *DevConn) Send(data []byte) error

Send 向连接写入一条文本消息。 并发安全由 gws.Conn.WriteMessage 保证；closed 仅作快速失败，与 WriteClose 竞态时以 gws 返回错误为准。

func (*DevConn) UpdateLast 

func (dc *DevConn) UpdateLast()

UpdateLast 更新连接最后活跃时间。原子写，无锁，便于消息循环中高频调用且不影响 CleanupExpired 遍历。

type ErrorHandler 

type ErrorHandler struct {
}

ErrorHandler WebSocket错误处理器（统一错误处理）

type Handle 

type Handle func(ctx context.Context, connCtx *ConnectionContext, body []byte) (interface{}, error)

核心类型定义 Handle 业务处理函数，返回 nil 则不回复。 对象池约束：*wire.JsonResp 由 replyData Put 回池；*wire.JsonBody 在 WebSocket 路径为每条消息的栈上/池对象（Process 的 jb），不得异步持有。 WS 业务 Handle 须在返回前结束对 bizData/connCtx 的使用；若启动 goroutine，不得捕获池化 JsonBody 指针。

type HeartbeatService 

type HeartbeatService struct {
	// contains filtered or unexported fields
}

HeartbeatService 心跳服务：gws 已内置 ping/pong，此服务仅用于定期清理过期连接

func NewHeartbeatService 

func NewHeartbeatService(interval, timeout time.Duration, manager *ConnectionManager) *HeartbeatService

func (*HeartbeatService) Start 

func (hs *HeartbeatService) Start()

func (*HeartbeatService) Stop 

func (hs *HeartbeatService) Stop()

type MessageHandler 

type MessageHandler struct {
	// contains filtered or unexported fields
}

MessageHandler 消息处理器：统一处理消息校验、解码、路由

func GetMessageHandler 

func GetMessageHandler(hook fwsign.CipherHook, handle Handle) *MessageHandler

func (*MessageHandler) CheckOuterSign 

func (self *MessageHandler) CheckOuterSign(usr int64, msg, sign []byte) (crypto.Cipher, error)

CheckOuterSign 按用户 ID 取 ML-DSA Cipher 并校验外层签名。

func (*MessageHandler) Process 

func (mh *MessageHandler) Process(connCtx *ConnectionContext, body []byte, jb *wire.JsonBody) (crypto.Cipher, interface{}, error)

Process 处理单条 WS 文本帧。jb 为本条消息独占的 JsonBody（由调用方从池取出并在调用结束后 Put），不得与 connCtx 共享指针以免并发覆盖。

type RouteInfo 

type RouteInfo struct {
	Handle       Handle        // 业务处理器
	RouterConfig *RouterConfig // 路由配置
}

RouteInfo WebSocket路由信息结构体

type RouterConfig 

type RouterConfig struct {
	Guest       bool
	UsePlan2    bool
	KeyRoute    bool
	LoginRoute  bool
	AesRequest  bool
	AesResponse bool
}

type WebSocketMetrics 

type WebSocketMetrics struct {
	// contains filtered or unexported fields
}

WebSocketMetrics WebSocket监控指标

type WsConnCloseHook 

type WsConnCloseHook func(ctx context.Context, connCtx *ConnectionContext, closeErr error)

WsConnCloseHook 在连接已从 ConnectionManager 移除之后、取消连接 context 之前执行。 closeErr 为 gws 关闭原因；若同 subject 仍有其它活跃连接，业务侧应自行判断是否标记离线。

type WsEventHandler 

type WsEventHandler struct {
	// contains filtered or unexported fields
}

WsEventHandler gws 事件处理器：处理 WebSocket 连接的生命周期事件

func (*WsEventHandler) OnClose 

func (h *WsEventHandler) OnClose(socket *gws.Conn, err error)

OnClose 连接关闭时的回调

func (*WsEventHandler) OnMessage 

func (h *WsEventHandler) OnMessage(socket *gws.Conn, message *gws.Message)

OnMessage 收到消息时的回调

func (*WsEventHandler) OnOpen 

func (h *WsEventHandler) OnOpen(socket *gws.Conn)

OnOpen 连接建立时的回调

func (*WsEventHandler) OnPing 

func (h *WsEventHandler) OnPing(socket *gws.Conn, payload []byte)

OnPing 收到 Ping 帧时的回调（gws 会自动回复 Pong）

func (*WsEventHandler) OnPong 

func (h *WsEventHandler) OnPong(socket *gws.Conn, payload []byte)

OnPong 收到 Pong 帧时的回调（gws 接口要求）

type WsHeartbeatHook 

type WsHeartbeatHook func(ctx context.Context, connCtx *ConnectionContext, jb *wire.JsonBody)

WsHeartbeatHook 在 /ws/ping 校验通过并更新连接 Last 之后执行；不进入业务 Pre/Post 链。 用于在线态刷新等副作用；hook 内应快速返回并自行异步落库，避免阻塞消息循环。

type WsMessagePostFilter 

type WsMessagePostFilter func(ctx context.Context, connCtx *ConnectionContext, jb *wire.JsonBody, bizData []byte, reply interface{}, handleErr error) error

WsMessagePostFilter 在业务 Handle 返回之后、构造成功回包（replyData）之前执行。 handleErr 为 Handle 的 error；仅做日志时返回 nil。若 handleErr 为 nil 且本函数返回非 nil，则本条按错误处理。 handleErr 非 nil 时，本函数返回值被忽略。使用约定：仅在监听前 AddWsPostFilter。

type WsMessagePreFilter 

type WsMessagePreFilter func(ctx context.Context, connCtx *ConnectionContext, jb *wire.JsonBody, bizData []byte) error

WsMessagePreFilter 在单条 WS 业务消息已完成解密且路由命中之后、调用业务 Handle 之前执行。 返回非 nil 时终止本条处理（与 Handle 返回错误一致，不发送成功回包）。 使用约定：仅在 StartWebsocket 监听前 AddWsPreFilter；/ws/ping 与未命中路由不进入此前置链。

type WsServer 

type WsServer struct {
	RedisCacheAware func(ds ...string) (cache.Cache, error) // 8字节 - 函数指针
	LocalCacheAware func(ds ...string) (cache.Cache, error) // 8字节 - 函数指针
	PushKeyProvider func(subject string) string             // 推送签名密钥获取函数：subject=="" 表示全量广播
	// contains filtered or unexported fields
}

WsServer WebSocket服务器核心结构体

func NewWsServer 

func NewWsServer(connUniquenessMode ConnectionUniquenessMode) *WsServer

NewWsServer 创建WebSocket服务器 connUniquenessMode: 连接唯一性模式

  • SubjectUnique: 一个用户只能有一个连接（适用于单设备应用）
  • SubjectDeviceUnique: 一个用户可以在多个设备上连接（适用于多设备场景）

func (*WsServer) AddCipherHook 

func (self *WsServer) AddCipherHook(hook fwsign.CipherHook) error

AddCipherHook 注册 Plan2 Cipher 动态加载回调（与 HttpNode.AddCipherHook 一致）。

func (*WsServer) AddJwtConfig 

func (self *WsServer) AddJwtConfig(config jwt.JwtConfig) error

AddJwtConfig 添加JWT配置

func (*WsServer) AddLocalCache 

func (self *WsServer) AddLocalCache(cacheAware CacheAware)

AddLocalCache 增加本地缓存实例

func (*WsServer) AddRedisCache 

func (self *WsServer) AddRedisCache(cacheAware CacheAware)

AddRedisCache 增加redis缓存实例

func (*WsServer) AddRouter 

func (s *WsServer) AddRouter(path string, handle Handle, routerConfig *RouterConfig) error

AddRouter 注册路由：path -> Handle。应在 StartWebsocket 之前完成所有注册。 不支持在服务启动后动态添加：消息循环中会无锁读 s.routes，动态添加会产生并发读写风险。

func (*WsServer) AddWsConnCloseHook 

func (s *WsServer) AddWsConnCloseHook(f WsConnCloseHook)

AddWsConnCloseHook 注册连接关闭钩子（ConnectionManager 移除之后执行）。 使用约定：仅在 StartWebsocket 监听前调用。

func (*WsServer) AddWsHeartbeatHook 

func (s *WsServer) AddWsHeartbeatHook(f WsHeartbeatHook)

AddWsHeartbeatHook 注册 /ws/ping 心跳钩子（校验通过后、回包前执行）。 使用约定：仅在 StartWebsocket 监听前调用。

func (*WsServer) AddWsPostFilter 

func (s *WsServer) AddWsPostFilter(f WsMessagePostFilter)

AddWsPostFilter 注册 WebSocket 业务后置过滤器（在 Handle 返回之后、replyData 之前执行）。 使用约定：仅在 StartWebsocket 监听前调用。

func (*WsServer) AddWsPreFilter 

func (s *WsServer) AddWsPreFilter(f WsMessagePreFilter)

AddWsPreFilter 注册 WebSocket 业务前置过滤器（在解密完成且路由命中之后、Handle 之前执行）。 使用约定：仅在 StartWebsocket 监听前调用；监听运行期间不得再调用（与 HttpNode.AddFilter 一致）。

func (*WsServer) BuildPlan2KeyResponse 

func (s *WsServer) BuildPlan2KeyResponse(req *wire.PublicKey) (*wire.PublicKey, error)

BuildPlan2KeyResponse 处理 plan2 的 /key 交换（ML-KEM + ML-DSA）。

func (*WsServer) CheckOuterSign 

func (s *WsServer) CheckOuterSign(cipher crypto.Cipher, msg, sign []byte) (crypto.Cipher, error)

func (*WsServer) GetCacheObject 

func (self *WsServer) GetCacheObject() (cache.Cache, error)

GetCacheObject 获取缓存对象

func (*WsServer) GetConnManager 

func (s *WsServer) GetConnManager() *ConnectionManager

GetConnManager 获取连接管理器（用于测试）

func (*WsServer) GetConnectionManager 

func (s *WsServer) GetConnectionManager() *ConnectionManager

GetConnectionManager 获取连接管理器（用于健康检查等操作）

func (*WsServer) GetMetrics 

func (s *WsServer) GetMetrics() *WebSocketMetrics

GetMetrics 获取监控指标快照（所有计数器均用 atomic 读取）

func (*WsServer) HTTPServer 

func (s *WsServer) HTTPServer() *http.Server

HTTPServer 返回内部标准 HTTP 服务器（启动 ListenAndServe 前可用于包装 Handler）。

func (*WsServer) LogMetrics 

func (s *WsServer) LogMetrics()

LogMetrics 记录当前监控指标到日志

func (*WsServer) NewPool 

func (s *WsServer) NewPool(maxConn, limit, bucket, ping int) error

maxConn = 300 // 允许的最大并发连接数 limit = 20 // 每秒允许的平均消息数（令牌桶速率） bucket = 100 // 令牌桶容量（突发消息缓冲） ping = 15 // 心跳间隔（秒）

func (*WsServer) NewPoolWithIdleTimeout 

func (s *WsServer) NewPoolWithIdleTimeout(maxConn, limit, bucket, ping, idleTimeoutSeconds int) error

NewPoolWithIdleTimeout 创建连接池并设置心跳周期与空闲踢线时间（秒）。 idleTimeoutSeconds <= 0 时回退为 2 倍 ping。

func (*WsServer) SetAuthTimeWindowSeconds 

func (s *WsServer) SetAuthTimeWindowSeconds(seconds int64)

SetAuthTimeWindowSeconds 设置请求时间窗（秒）。<=0 时忽略。

func (*WsServer) SetIdleTimeout 

func (s *WsServer) SetIdleTimeout(timeout time.Duration)

SetIdleTimeout 设置连接空闲超时时间

func (*WsServer) SetMaxBodyLen 

func (s *WsServer) SetMaxBodyLen(n int)

SetMaxBodyLen 设置单条消息体最大长度（字节），需在 StartWebsocket 前调用

func (*WsServer) SetParallelEnabled 

func (s *WsServer) SetParallelEnabled(enabled bool)

SetParallelEnabled 设置是否并行处理同一连接上的消息（需在 StartWebsocket 前调用）。 true 可提升吞吐；false 可保证单连接消息按处理顺序串行。

func (*WsServer) SetPlan2SharedKeyTTLSeconds 

func (s *WsServer) SetPlan2SharedKeyTTLSeconds(seconds int64)

SetPlan2SharedKeyTTLSeconds 设置 plan2 临时共享密钥 TTL（秒）。<=0 时忽略。

func (*WsServer) SetPushKeyProvider 

func (s *WsServer) SetPushKeyProvider(provider func(subject string) string)

SetPushKeyProvider 设置推送签名密钥获取函数。subject=="" 表示全量广播。

func (*WsServer) SetValidateTokenPerMessage 

func (s *WsServer) SetValidateTokenPerMessage(validate bool)

SetValidateTokenPerMessage 设置是否在每条消息时校验 token 有效期（validWebSocketBody 内生效）。 false（默认）：仅建连时校验，连接期间 token 过期不踢线，性能更好； true：每条消息校验 exp，过期即 401，适合强安全/合规场景。

func (*WsServer) StartWebsocket 

func (s *WsServer) StartWebsocket(addr string) error

func (*WsServer) StopWebsocket 

func (s *WsServer) StopWebsocket() (err error)

StopWebsocket 停止WebSocket服务器

func (*WsServer) StopWebsocketWithTimeout 

func (s *WsServer) StopWebsocketWithTimeout(timeout time.Duration) error

StopWebsocketWithTimeout 带超时的优雅关闭

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.