ws

package
v0.6.4 Latest Latest
Warning

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

 Go to latest Published: Jul 19, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	MsgTypeAuth      = "auth"
	MsgTypeError     = "error"
	MsgTypeClose     = "close"
	MsgTypeBroadcast = "broadcast"
	MsgTypeAuthOK    = "auth_ok"
)

系统消息类型

Variables

View Source 
var (
	ErrAuthFailed       = errors.New("鉴权失败")
	ErrAuthTimeout      = errors.New("鉴权超时，断开连接")
	ErrConnectionClosed = errors.New("连接已关闭")
	ErrMessageQueueFull = errors.New("消息队列已满")
	ErrInvalidMessage   = errors.New("无效消息格式")
	ErrClientNotFound   = errors.New("客户端不存在")
	ErrHubClosed        = errors.New("Hub 已关闭")
	ErrHandlerNotFound  = errors.New("处理器不存在")
)

错误定义

Functions

func ExampleUsage 

func ExampleUsage()

ExampleUsage 使用示例

func ExampleWithJWTAuth 

func ExampleWithJWTAuth()

简单的 JWT 风格的 token 验证示例

Types

type AuthHandler 

type AuthHandler func(message Message) (clientID string, err error)

回调函数类型定义

type Client 

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

Client WebSocket 客户端实现

func (*Client) GetMetadata 

func (c *Client) GetMetadata() map[string]any

GetMetadata 返回客户端元数据的副本，修改返回值不影响内部状态

func (*Client) ID 

func (c *Client) ID() string

ID 返回客户端 ID：连接建立时为 UUID，鉴权成功后被业务 ID 覆盖（鉴权回调返回空 ID 则保留 UUID）

func (*Client) IsAuthenticated 

func (c *Client) IsAuthenticated() bool

IsAuthenticated 检查客户端是否已通过鉴权

func (*Client) JoinGroup 

func (c *Client) JoinGroup(groupID string)

JoinGroup 将客户端加入指定分组（房间），重复加入幂等，分组不存在时隐式创建。 组名为空时直接忽略；连接断开后由 Hub 自动清出所有分组，无需手动退组。

func (*Client) LeaveGroup 

func (c *Client) LeaveGroup(groupID string)

LeaveGroup 将客户端移出指定分组，不在分组中时为空操作。

func (*Client) Request 

func (c *Client) Request() *http.Request

Request 返回客户端升级时的原始 HTTP 请求，业务 handler 可用它拼接完整 URL 等。 测试或没有真实 HTTP 请求的场景可能返回 nil，调用方需自行判空。

func (*Client) Send 

func (c *Client) Send(ctx context.Context, message Message) error

Send 将消息投入发送队列，队列满时阻塞等待，直至队列可用、连接关闭或 ctx 超时。

func (*Client) SetMetadata 

func (c *Client) SetMetadata(key string, value any)

SetMetadata 设置客户端级元数据，供业务 handler 存取连接上下文

type Config 

type Config struct {
	// ReadBufferSize 底层连接的读缓冲区大小（字节），影响升级后每次系统调用的读取量
	ReadBufferSize int
	// WriteBufferSize 底层连接的写缓冲区大小（字节）
	WriteBufferSize int
	// MaxMessageSize 单条入站消息的最大字节数，超过则断开连接，防止大消息耗尽内存
	MaxMessageSize int64
	// WriteTimeout 单次写操作的超时时间，超时未写完视为连接异常
	WriteTimeout time.Duration
	// HeartbeatInterval 服务端向客户端发送 Ping 帧的间隔
	HeartbeatInterval time.Duration
	// HeartbeatTimeout 读超时时间，超过此时间未收到客户端任何数据（含 Pong）则判定连接死亡
	HeartbeatTimeout time.Duration
	// AuthTimeout 连接建立后等待鉴权的最长时间，超时未鉴权则主动断开
	AuthTimeout time.Duration
	// MaxConnections Hub 允许的最大并发连接数，超限的新连接会被拒绝
	MaxConnections int
	// MessageQueueSize 每个客户端发送队列的长度，队列满时发送方阻塞等待，直至 ctx 超时或连接关闭
	MessageQueueSize int
	// EventQueueSize Hub 内部事件通道（注册、注销、鉴权登记、广播）共用的缓冲长度
	EventQueueSize int
	// SendToClientQueueSize 定向发送请求通道的缓冲长度，写满后 SendToClient 阻塞等待直至 ctx 超时
	SendToClientQueueSize int
	// GetClientsQueueSize 获取客户端列表请求通道的缓冲长度，写满后 GetClients 阻塞等待直至 ctx 超时
	GetClientsQueueSize int
	// EnableCompression 是否启用 WebSocket 压缩扩展（permessage-deflate），会消耗少量 CPU
	EnableCompression bool
	// CheckOrigin 跨域校验函数，返回 false 则拒绝升级；默认放行所有来源
	CheckOrigin func(r *http.Request) bool
}

Config 配置选项

func DefaultConfig 

func DefaultConfig() *Config

DefaultConfig 返回默认的 WebSocket 配置

type ConfigOption 

type ConfigOption func(*Config)

ConfigOption 配置覆写函数，经 NewHub 传入以修改默认配置

type ConnectHandler 

type ConnectHandler func(client *Client) error

回调函数类型定义

type DisconnectHandler 

type DisconnectHandler func(client *Client, err error)

回调函数类型定义

type ErrorHandler 

type ErrorHandler func(client *Client, err error)

回调函数类型定义

type ErrorMessage 

type ErrorMessage struct {
	MsgType string `json:"type"`
	Msg     string `json:"msg"`
}

ErrorMessage 错误消息实现

func NewErrorMessage 

func NewErrorMessage(msg string) *ErrorMessage

NewErrorMessage 创建新的错误消息

func (*ErrorMessage) Data 

func (e *ErrorMessage) Data() json.RawMessage

Data 返回 nil，错误消息无业务数据

func (*ErrorMessage) Marshal 

func (e *ErrorMessage) Marshal() ([]byte, error)

Marshal 将错误消息序列化为 JSON 字节数组

func (*ErrorMessage) Type 

func (e *ErrorMessage) Type() string

Type 返回错误消息类型（恒为 MsgTypeError）

type ExampleChatServer 

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

ExampleChatServer 示例聊天服务器

func NewExampleChatServer 

func NewExampleChatServer() *ExampleChatServer

NewExampleChatServer 创建示例聊天服务器

func (*ExampleChatServer) Close 

func (s *ExampleChatServer) Close()

Close 关闭服务器

func (*ExampleChatServer) ServeHTTP 

func (s *ExampleChatServer) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP 处理 WebSocket 连接

type Handler 

type Handler interface {
	Handle(client *Client, message Message) error
}

Handler 消息处理器接口

func Wrap 

func Wrap[T any](fn func(client *Client, data T) error) Handler

Wrap 将强类型业务函数包装为 Handler：消息数据绑定到 T 后再调用 fn， 业务方直接操作具名字段，无需对 map 取值做类型断言。 字段类型不符时返回反序列化错误，由框架统一回错。

type HandlerFunc 

type HandlerFunc func(client *Client, message Message) error

HandlerFunc 函数类型实现 Handler 接口

func (HandlerFunc) Handle 

func (f HandlerFunc) Handle(client *Client, message Message) error

Handle 实现 Handler 接口，直接调用函数本体

type Hub 

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

Hub WebSocket 连接管理中心实现

func NewHub 

func NewHub(opt ...ConfigOption) *Hub

NewHub 创建新的 WebSocket Hub 实例

func (*Hub) Broadcast 

func (h *Hub) Broadcast(message Message)

Broadcast 向所有已认证的客户端广播消息；广播通道满时丢弃并记录告警日志

func (*Hub) Close 

func (h *Hub) Close()

Close 关闭 Hub 并断开所有客户端连接，幂等

func (*Hub) CloseClient 

func (h *Hub) CloseClient(clientID string) error

CloseClient 按 ID 踢下线：关闭该 ID 下所有连接（多标签页一并断开），阻塞等待 run() 处理完毕。 ID 不存在时目标态已达成，按幂等返回 nil；Hub 关闭时返回 ErrHubClosed。

func (*Hub) GetClients 

func (h *Hub) GetClients() []*Client

GetClients 返回所有已认证客户端的列表快照；Hub 关闭时返回空列表

func (*Hub) GroupSize 

func (h *Hub) GroupSize(groupID string) int

GroupSize 获取指定分组内的连接数；分组不存在返回 0，Hub 关闭返回 0。

func (*Hub) Handle 

func (h *Hub) Handle(msgType string, handler Handler)

Handle 注册指定类型的业务消息处理器

func (*Hub) SendToClient 

func (h *Hub) SendToClient(ctx context.Context, clientID string, message Message) error

SendToClient 同步向指定客户端发送消息，阻塞等待 run() 处理完毕并返回投递结果。 ctx 控制入队与等待响应的最长时间；Hub 关闭时立即返回 ErrHubClosed。

func (*Hub) SendToClientAsync 

func (h *Hub) SendToClientAsync(ctx context.Context, clientID string, message Message) error

SendToClientAsync 异步向指定客户端发送消息：等待请求入队即返回，不等待投递结果。 返回值仅表示入队是否成功（Hub 关闭或 ctx 超时），不代表消息是否送达。

func (*Hub) SendToGroup 

func (h *Hub) SendToGroup(ctx context.Context, groupID string, message Message) error

SendToGroup 同步向指定分组内所有客户端发送消息，阻塞等待 run() 投递完毕。 队列积压的慢连接跳过并记录告警日志；ctx 控制入队与等待响应的最长时间； Hub 关闭时立即返回 ErrHubClosed。

func (*Hub) SendToGroupAsync 

func (h *Hub) SendToGroupAsync(ctx context.Context, groupID string, message Message) error

SendToGroupAsync 异步向指定分组发送消息：等待请求入队即返回，不等待投递结果。 返回值仅表示入队是否成功（Hub 关闭或 ctx 超时），不代表消息是否送达。

func (*Hub) ServeHTTP 

func (h *Hub) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP 实现 http.Handler，处理 WebSocket 升级并将新连接注册到 Hub

func (*Hub) SetAuthHandler 

func (h *Hub) SetAuthHandler(handler AuthHandler)

SetAuthHandler type=auth

func (*Hub) SetConnectHandler 

func (h *Hub) SetConnectHandler(handler ConnectHandler)

SetConnectHandler 设置客户端连接处理器，新连接注册后于独立 goroutine 触发； 触发时连接尚未鉴权，ID 为 UUID

func (*Hub) SetDefaultHandler 

func (h *Hub) SetDefaultHandler(handler Handler)

SetDefaultHandler 设置默认消息处理器，处理未注册类型的消息

func (*Hub) SetDisconnectHandler 

func (h *Hub) SetDisconnectHandler(handler DisconnectHandler)

SetDisconnectHandler 设置客户端断开连接处理器，连接除名后于独立 goroutine 触发

func (*Hub) SetErrorHandler 

func (h *Hub) SetErrorHandler(handler ErrorHandler)

SetErrorHandler 设置错误处理器，消息解析失败、处理器报错、未鉴权发业务消息时触发

type Huber 

type Huber interface {
	// ServeHTTP 处理 HTTP 升级为 WebSocket 的请求
	ServeHTTP(w http.ResponseWriter, r *http.Request)
	// Broadcast 向所有已认证的客户端广播消息
	Broadcast(message Message)
	// SendToClient 向指定客户端发送消息，阻塞等待投递结果，ctx 控制最长等待时间
	SendToClient(ctx context.Context, clientID string, message Message) error
	// SendToClientAsync 向指定客户端发送消息，仅等待入队成功，不等待投递结果
	SendToClientAsync(ctx context.Context, clientID string, message Message) error
	// CloseClient 按 ID 踢下线：关闭该 ID 下所有连接（多标签页一并断开），阻塞等待处理完毕。
	// ID 不存在时目标态已达成，按幂等返回成功；断开走正常 leave 流程，触发 disconnect 回调
	CloseClient(clientID string) error
	// SendToGroup 向指定分组内所有客户端发送消息，阻塞等待投递完毕，ctx 控制最长等待时间。
	// 队列积压的慢连接跳过并记录告警日志，不剔除连接；分组不存在或为空时直接返回成功
	SendToGroup(ctx context.Context, groupID string, message Message) error
	// SendToGroupAsync 向指定分组发送消息，仅等待入队成功，不等待投递结果
	SendToGroupAsync(ctx context.Context, groupID string, message Message) error
	// GroupSize 获取指定分组内的连接数，分组不存在时返回 0
	GroupSize(groupID string) int
	// GetClients 获取所有已认证的客户端列表
	GetClients() []*Client
	// Close 关闭 Hub 并清理所有资源
	Close()
	// SetAuthHandler 设置鉴权处理器。
	// 处理器返回的 clientID 必须唯一标识业务主体：同一 ID 的多个连接视为同一主体（多标签页），
	// 不同主体共用一个 ID 会导致 SendToClient 串号投递。返回空 ID 时保留连接建立时分配的 UUID。
	SetAuthHandler(handler AuthHandler)
	// SetConnectHandler 设置客户端连接处理器
	SetConnectHandler(handler ConnectHandler)
	// SetDisconnectHandler 设置客户端断开连接处理器
	SetDisconnectHandler(handler DisconnectHandler)
	// SetErrorHandler 设置错误处理器
	SetErrorHandler(handler ErrorHandler)

	// Handle 注册指定类型的消息处理器，用法同 http.Handle
	Handle(msgType string, handler Handler)
	// SetDefaultHandler 设置默认消息处理器（处理未注册类型的消息）
	SetDefaultHandler(handler Handler)
}

Huber WebSocket 连接管理中心接口

type Message 

type Message interface {
	// Type 获取消息类型
	Type() string
	// Data 获取消息数据的原始 JSON 字节，无数据时返回 nil
	Data() json.RawMessage
	// Marshal 将消息序列化为 JSON 字节数组
	Marshal() ([]byte, error)
}

Message 消息接口。 实现必须是只读的：广播时同一实例被分发到所有连接并并发调用 Marshal， 携带可变状态的实现会产生数据竞争。

type MessageRouter 

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

MessageRouter 消息路由器，负责管理消息处理器

func NewMessageRouter 

func NewMessageRouter() *MessageRouter

NewMessageRouter 创建新的消息路由器实例

func (*MessageRouter) GetHandler 

func (r *MessageRouter) GetHandler(msgType string) Handler

GetHandler 获取指定消息类型的处理器，如果不存在则返回默认处理器

func (*MessageRouter) RegisterHandler 

func (r *MessageRouter) RegisterHandler(msgType string, handler Handler)

RegisterHandler 为指定消息类型注册处理器

func (*MessageRouter) SetDefaultHandler 

func (r *MessageRouter) SetDefaultHandler(handler Handler)

SetDefaultHandler 设置默认消息处理器

type StandardMessage 

type StandardMessage struct {
	MsgType string `json:"type"`
	Payload any    `json:"data,omitempty"`
	// contains filtered or unexported fields
}

StandardMessage 标准消息实现

func NewMessage 

func NewMessage(msgType string, data any) *StandardMessage

NewMessage 创建新的标准消息

func (*StandardMessage) Data 

func (m *StandardMessage) Data() json.RawMessage

Data 返回消息数据的原始 JSON 字节。 入站消息直返所存原始字节；出站消息将 Payload 序列化为字节。 每次调用独立计算，不缓存、不改写自身状态，满足并发只读约束。

func (*StandardMessage) Marshal 

func (m *StandardMessage) Marshal() ([]byte, error)

Marshal 将消息序列化为 JSON 字节数组。 结果以 sync.Once 缓存：同一实例多次调用仅序列化一次， 广播分发场景下 N 个连接共享一趟序列化。 首次调用后 Payload 不可再改，否则缓存与实际内容脱节（消息合约本即只读）。

func (*StandardMessage) Type 

func (m *StandardMessage) Type() string

Type 返回消息类型

func (*StandardMessage) UnmarshalJSON 

func (m *StandardMessage) UnmarshalJSON(data []byte) error

UnmarshalJSON 实现 json.Unmarshaler，反序列化错误由基础库原样抛出。 内容被改写后序列化缓存随之失效，下次 Marshal 重新计算。 经别名类型绕行，避免 json.Unmarshal 递归调回本方法。

Source Files

View all Source files

Directories

Path Synopsis

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.