Documentation
¶
Overview ¶
Package message 提供基于自定义二进制协议的消息类型、工厂注册和连接封装。
本包使用固定 4 字节头部编码消息:前 2 字节为 MessageType,后 2 字节为 payload 长度, 因此单条消息的 payload 最大为 uint16 上限。调用方可以通过 Message、MessageFactory 和 FactoryRegister 扩展自定义消息类型;内置实现提供心跳消息、单字符串消息以及与该协议配套的 Scanner。
WrapConn 会把 net.Conn 包装为按上述协议收发消息的连接,并可按给定间隔发送心跳包。 连接上的并发、生命周期和共享 channel 约束以 Conn 及其方法文档为准。
Index ¶
- func FactoryRegister(messageType MessageType, messageFunc GenerateMessageFunc) error
- func NewHeartbeatMessage(serialNumber uint64) *heartbeatMessage
- func NewMessageFactory() *messageFactory
- func NewScanner(r io.Reader) *bufio.Scanner
- func NewSingleStringMessage(message string) *singleStringMessage
- func WrapConn(c net.Conn, heartbeatInterval time.Duration) *conn
- type Conn
- type GenerateMessageFunc
- type Generator
- type HeartbeatMessage
- type Message
- type MessageFactory
- type MessageType
- type SingleStringMessage
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func FactoryRegister ¶
func FactoryRegister(messageType MessageType, messageFunc GenerateMessageFunc) error
FactoryRegister 将消息类型注册到默认工厂。
参数:
- messageType: 待注册的消息类型。
- messageFunc: 对应的消息生成函数。
返回:
- error: 默认工厂注册失败时返回错误。
func NewHeartbeatMessage ¶
func NewHeartbeatMessage(serialNumber uint64) *heartbeatMessage
NewHeartbeatMessage 创建心跳消息。
参数:
- serialNumber: 要写入消息的心跳序列号。
返回:
- *heartbeatMessage: 新创建的心跳消息实例。
func NewMessageFactory ¶
func NewMessageFactory() *messageFactory
NewMessageFactory 创建新的消息工厂实例。
返回的工厂可并发调用 Register;Generate 依赖底层 map 读取,通常应在完成注册后再供并发生成使用。
参数:无。
返回:
- *messageFactory: 新创建的消息工厂实例。
func NewScanner ¶
NewScanner 创建按本包协议拆分消息包的 bufio.Scanner。
返回的 Scanner 使用 [scanMessage] 作为 SplitFunc,并将 token 缓冲区上限设置为协议允许的最大完整包长度。
参数:
- r: 提供协议字节流的输入源。
返回:
- *bufio.Scanner: 按本包协议拆分消息包的 Scanner。
func NewSingleStringMessage ¶
func NewSingleStringMessage(message string) *singleStringMessage
NewSingleStringMessage 创建单字符串消息。
参数:
- message: 要写入消息 payload 的字符串内容。
返回:
- *singleStringMessage: 新创建的单字符串消息实例。
func WrapConn ¶
WrapConn 将底层 net.Conn 包装为按本包协议异步收发消息的连接。
返回的连接会创建容量为 5120 的接收与发送队列,但不会自动启动后台任务; 调用方需要显式调用 Conn.Start 启动读写循环,且 Start 只应调用一次。 heartbeatInterval 大于 0 时,Start 会额外提交定时心跳发送任务。
参数:
- c: 待包装的底层网络连接,必须非 nil;调用方负责保证其满足所需的 net.Conn 语义,传入 nil 会导致后续使用时 panic。
- heartbeatInterval: 心跳发送间隔;小于等于 0 时不会启动心跳任务。
返回:
- *conn: 包装后的协议连接实例,初始处于未关闭状态;调用方应在不再使用时调用 Close。
Types ¶
type Conn ¶
type Conn interface {
// Close 关闭连接并通知内部 goroutine 退出。
//
// Close 可与 Closed 和 SendMessage 并发调用;重复调用只会在首次关闭时关闭底层连接,
// 并关闭 [Conn.Message] 返回的共享接收 channel。
//
// 参数:无。
//
// 返回:
// - error: 首次关闭底层连接时返回的错误;连接已关闭时返回 nil。
Close() error
// LocalAddr 返回底层连接的本地网络地址。
//
// 参数:无。
//
// 返回:
// - net.Addr: 本地网络地址。
LocalAddr() net.Addr
// RemoteAddr 返回底层连接的远程网络地址。
//
// 参数:无。
//
// 返回:
// - net.Addr: 远程网络地址。
RemoteAddr() net.Addr
// SetDeadline 设置底层连接的读写截止时间。
//
// 调用后会同时影响 [Conn.Start] 启动的内部收发流程。
//
// 参数:
// - time.Time: 截止时间;零值表示取消已设置的读写截止时间。
//
// 返回:
// - error: 底层连接设置截止时间失败时返回错误。
SetDeadline(time.Time) error
// SetReadDeadline 设置底层连接的读截止时间。
//
// 该设置会影响 [Conn.Start] 启动的内部接收流程。
//
// 参数:
// - time.Time: 截止时间;零值表示取消已设置的读截止时间。
//
// 返回:
// - error: 底层连接设置读截止时间失败时返回错误。
SetReadDeadline(time.Time) error
// SetWriteDeadline 设置底层连接的写截止时间。
//
// 该设置会影响 [Conn.Start] 启动的内部发送流程。
//
// 参数:
// - time.Time: 截止时间;零值表示取消已设置的写截止时间。
//
// 返回:
// - error: 底层连接设置写截止时间失败时返回错误。
SetWriteDeadline(time.Time) error
// Closed 返回连接是否已经关闭。
//
// Closed 可与 Close 和 SendMessage 并发调用。
//
// 参数:无。
//
// 返回:
// - bool: 连接是否已经进入关闭状态。
Closed() bool
// Start 提交内部发送、接收以及可选心跳任务。
//
// Start 不会自行去重,调用方只应调用一次。传入的上下文结束或连接关闭后,
// 已成功启动的内部任务会退出。任务提交通过包级 goroutine 池完成,提交失败时
// 当前签名不会向调用方返回错误。
//
// 参数:
// - context.Context: 控制内部 goroutine 生命周期的上下文,不能为空。
Start(context.Context)
// SendMessage 将消息放入内部发送队列。
//
// SendMessage 可与 Close 和 Closed 并发调用。返回 nil 仅表示消息已入队,
// 不表示已经写入底层连接;当发送队列已满时会阻塞,直到队列腾出空间或连接关闭。
//
// 参数:
// - Message: 待异步发送的消息;调用方应保证其非 nil。
//
// 返回:
// - error: 连接已关闭,或消息在入队前因收到关闭通知而被拒绝时返回错误。
SendMessage(Message) error
// Message 返回连接的共享接收 channel。
//
// 该 channel 只创建一次;多个消费者同时读取时会竞争消费消息。
// 连接关闭时,该 channel 也会被关闭。
//
// 参数:无。
//
// 返回:
// - <-chan Message: 共享的只读消息通道。
Message() <-chan Message
}
Conn 定义按本包协议收发消息的连接契约。
Close、Closed 和 SendMessage 可以并发调用。Start 不会自行去重, 调用方只应调用一次;重复调用会重复提交内部收发任务和可选心跳任务, 并竞争同一底层连接。Message 返回单个共享接收 channel,多个消费者 同时读取时会竞争消费消息;连接关闭时该 channel 会被关闭。 地址查询方法直接委托给底层 net.Conn;截止时间设置会同步影响 Conn.Start 启动的内部收发流程。
type GenerateMessageFunc ¶
type GenerateMessageFunc func(MessageType, []byte) (Message, error)
GenerateMessageFunc 适配普通函数为 Generator 实现。
适配函数收到的 payload 不包含协议头中的 MessageType 和长度字段。
参数:
- MessageType: 目标消息类型。
- []byte: 待解码的消息 payload。
返回:
- Message: 生成的消息实例。
- error: 生成失败时返回错误。
func (GenerateMessageFunc) GenerateMessage ¶
func (f GenerateMessageFunc) GenerateMessage(messageType MessageType, payload []byte) (Message, error)
GenerateMessage 调用底层函数生成消息实例。
参数:
- messageType: 目标消息类型。
- payload: 待解码的消息 payload。
返回:
- Message: 生成的消息实例。
- error: 底层适配函数返回的错误。
type Generator ¶
type Generator interface {
// GenerateMessage 根据消息类型和 payload 生成消息实例。
//
// payload 不包含协议头中的 MessageType 和长度字段。
//
// 参数:
// - messageType: 目标消息类型。
// - payload: 待解码的消息 payload。
//
// 返回:
// - Message: 生成的消息实例。
// - error: 生成失败时返回错误。
GenerateMessage(messageType MessageType, payload []byte) (Message, error)
}
Generator 根据消息类型和 payload 生成具体消息实例。
type HeartbeatMessage ¶
type HeartbeatMessage interface {
// SerialNumber 返回心跳消息中的序列号。
//
// 参数:无。
//
// 返回:
// - uint64: 心跳消息中的序列号。
SerialNumber() uint64
}
HeartbeatMessage 表示携带心跳序列号的消息。
type Message ¶
type Message interface {
// MessageType 返回消息的协议类型。
//
// 参数:无。
//
// 返回:
// - MessageType: 当前消息的协议类型。
MessageType() MessageType
// Pack 将消息编码为 payload。
//
// 返回结果不包含协议头中的 MessageType 和长度字段。
//
// 参数:无。
//
// 返回:
// - []byte: 当前消息编码后的 payload。
// - error: 编码失败时返回错误。
Pack() ([]byte, error)
// Unpack 使用 payload 还原消息内容。
//
// payload 不包含协议头中的 MessageType 和长度字段。
//
// 参数:
// - payload: 需要解码的消息 payload。
//
// 返回:
// - error: 解码失败时返回错误。
Unpack(payload []byte) error
}
Message 定义协议消息的类型、封包和解包契约。
Pack 和 Unpack 只处理 payload,不包含协议头中的 MessageType 和长度字段。
func FactoryGenerate ¶
func FactoryGenerate(messageType MessageType, payload []byte) (Message, error)
FactoryGenerate 使用默认工厂根据消息类型和 payload 创建消息实例。
参数:
- messageType: 目标消息类型。
- payload: 待解码的消息 payload。
返回:
- Message: 生成的消息实例。
- error: 默认工厂生成失败时返回错误。
func GenerateHeartbeatMessage ¶
func GenerateHeartbeatMessage(messageType MessageType, payload []byte) (Message, error)
GenerateHeartbeatMessage 根据消息类型和 payload 生成心跳消息。
messageType 必须等于 HeartbeatMessageType,payload 不能为空。
参数:
- messageType: 目标消息类型。
- payload: 待解码的心跳消息 payload。
返回:
- Message: 生成的心跳消息实例。
- error: messageType 不匹配、payload 为 nil 或解码失败时返回错误。
func GenerateSingleStringMessage ¶
func GenerateSingleStringMessage(messageType MessageType, payload []byte) (Message, error)
GenerateSingleStringMessage 根据消息类型和 payload 生成单字符串消息。
messageType 必须等于 SingleStringMessageType;nil payload 会被拒绝,非 nil 的空 payload 表示合法空字符串。
参数:
- messageType: 目标消息类型。
- payload: 待解码的字符串消息 payload。
返回:
- Message: 生成的单字符串消息实例。
- error: messageType 不匹配、payload 为 nil 或解码失败时返回错误。
type MessageFactory ¶
type MessageFactory interface {
// Register 将消息类型注册到工厂。
//
// 同一 messageType 只能注册一次;messageFunc 不能为空。
//
// 参数:
// - messageType: 待注册的消息类型。
// - messageFunc: 对应的消息生成函数。
//
// 返回:
// - error: 消息类型已存在或 messageFunc 为空时返回错误。
Register(messageType MessageType, messageFunc GenerateMessageFunc) error
// Generate 根据消息类型和 payload 创建消息实例。
//
// payload 不能为空,但可以是非 nil 的空切片。
//
// 参数:
// - messageType: 目标消息类型。
// - payload: 待解码的消息 payload。
//
// 返回:
// - Message: 生成的消息实例。
// - error: 消息类型未注册、payload 为 nil 或生成失败时返回错误。
Generate(messageType MessageType, payload []byte) (Message, error)
}
MessageFactory 定义消息类型注册和消息生成契约。
Register 负责将消息类型映射到生成函数;Generate 根据消息类型和 payload 还原消息实例。
type MessageType ¶
type MessageType uint16
MessageType 标识协议中的消息类型。
内置消息类型包括:
- HeartbeatMessageType: 心跳消息类型。
- SingleStringMessageType: 仅携带单个字符串 payload 的消息类型。
调用方可通过 FactoryRegister 注册其它 uint16 值作为自定义消息类型。
const ( // HeartbeatMessageType 表示心跳消息类型。 HeartbeatMessageType MessageType = 0x80 // SingleStringMessageType 表示仅携带单个字符串 payload 的消息类型。 SingleStringMessageType MessageType = 0x09 )
type SingleStringMessage ¶
type SingleStringMessage interface {
// Message 返回消息中的字符串内容。
//
// 参数:无。
//
// 返回:
// - string: 当前消息中的字符串内容。
Message() string
}
SingleStringMessage 表示仅携带一个字符串 payload 的消息。