diff --git a/examples/frame/main.go b/examples/frame/main.go new file mode 100644 index 0000000..04dd487 --- /dev/null +++ b/examples/frame/main.go @@ -0,0 +1,42 @@ +// frame 包示例:应用层帧协议(8 字节头 + 数据)在 net.Pipe 上的读写往返 +package main + +import ( + "fmt" + "net" + + "git.zeroonesoft.cn/golib/zogo/frame" +) + +func main() { + c1, c2 := net.Pipe() + defer c1.Close() + defer c2.Close() + + // 组帧:版本 + 命令字 + 流 ID + 数据 + f := frame.NewFrame(frame.FrameVersion, 0x10, 42) + if err := f.Marshal(map[string]any{"msg": "hello"}); err != nil { + fmt.Println("Marshal 失败:", err) + return + } + fmt.Println("帧信息:", f) + + // 写帧 / 读帧 + go func() { + if err := frame.WriteFrame(c1, f); err != nil { + fmt.Println("WriteFrame 失败:", err) + } + }() + + got, err := frame.ReadFrame(c2) + if err != nil { + fmt.Println("ReadFrame 失败:", err) + return + } + var data map[string]any + if err := got.Unmarshal(&data); err != nil { + fmt.Println("Unmarshal 失败:", err) + return + } + fmt.Printf("收到帧: %s 数据: %v\n", got.String(), data) +} diff --git a/frame/README.md b/frame/README.md new file mode 100644 index 0000000..3c0ec6d --- /dev/null +++ b/frame/README.md @@ -0,0 +1,50 @@ +# frame + +应用层帧协议:头 8 字节(版本 1 + 命令 1 + 长度 2 + 流 ID 4,小端序)+ 数据, +提供帧与 `net.Conn` 间的读写封装,配合 JSON 编解码使用。 + +> 迁移自 go-hua/common/frame.go,独立成包。 + +## 帧格式 + +``` ++---------+---------+-----------+-----------+----------------+ +| ver (1) | cmd (1) | length(2) | sid (4) | data (length) | ++---------+---------+-----------+-----------+----------------+ + 小端序,length 最大 65535 +``` + +## 用法 + +```go +import ( + "net" + + "git.zeroonesoft.cn/golib/zogo/frame" +) + +// 组帧:版本 + 命令字 + 流 ID,数据可 Marshal 任意 JSON 值 +f := frame.NewFrame(frame.FrameVersion, 0x10, 42) +f.Marshal(map[string]any{"msg": "hello"}) + +// 连接上写帧 / 读帧(阻塞直到完整帧) +frame.WriteFrame(conn, f) +got, err := frame.ReadFrame(conn) +var data map[string]any +got.Unmarshal(&data) +``` + +完整可运行例程:[examples/frame/main.go](../examples/frame/main.go) + +## 迁移修复(相对 go-hua/common 原实现) + +1. **SetData 超长未报错**:原实现 `fmt.Errorf` 结果未 return,超长数据静默截断写入; + 现超 65535 字节正确返回错误(有回归测试)。 +2. **ReadFrame 短读/吞字节**:原实现直接 `conn.Read` 读头(TCP 分段会错位),且用 + 一次性 `bufio.Reader` 读体(缓冲里多读到的后续帧字节随 reader 丢弃)。现统一改用 + `io.ReadFull` 精确读取,连续多帧不丢数据(有回归测试)。 + +## 注意 + +- 单帧数据上限 65535 字节(length 字段 2 字节),更大 payload 请业务侧分片。 +- `ReadFrame`/`WriteFrame` 非并发安全,同一连接的读与写可各由一个 goroutine 持有。 diff --git a/frame/frame.go b/frame/frame.go new file mode 100644 index 0000000..a932c17 --- /dev/null +++ b/frame/frame.go @@ -0,0 +1,172 @@ +// Package frame 应用层帧协议:头 8 字节(版本1 + 命令1 + 长度2 + 流ID4,小端序)+ 数据, +// 提供帧与 net.Conn 间的读写封装。 +// +// 迁移自 go-hua/common,迁移时修复两处问题: +// 1. SetData 超长时未返回错误(fmt.Errorf 结果未 return); +// 2. ReadFrame 原实现用临时 bufio.Reader,帧头短读会错位、 +// 一次 Peek 之外的缓冲字节会被丢弃(连续帧丢数据),现改用 io.ReadFull 精确读取。 +package frame + +import ( + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net" +) + +const FrameVersion byte = 1 + +// 帧头部 +const ( + sizeOfVer = 1 + sizeOfCmd = 1 + sizeOfLength = 2 + sizeOfSid = 4 + headerSize = sizeOfVer + sizeOfCmd + sizeOfSid + sizeOfLength +) + +const ( + frameBufSize = 0x10000 +) + +// Frame 应用层帧:头 8 字节(版本1 + 命令1 + 长度2 + 流ID4,小端序)+ 数据 +type Frame struct { + ver byte + cmd byte + sid uint32 + data []byte +} + +// NewFrame 创建帧,version 为协议版本,cmd 为命令字,sid 为流 ID +func NewFrame(version byte, cmd byte, sid uint32) Frame { + return Frame{ver: version, cmd: cmd, sid: sid} +} + +// ReadFrame 从连接读取一帧(阻塞直到完整帧读出;头/体不完整或连接中断返回错误) +func ReadFrame(conn net.Conn) (Frame, error) { + f := NewFrame(0, 0, 0) + hdr := RawHeader{} + // 读取头(io.ReadFull 防止头部被 TCP 分段导致的短读) + if _, err := io.ReadFull(conn, hdr[:]); err != nil { + return f, err + } + + f.ver = hdr.Version() + f.cmd = hdr.Cmd() + f.sid = hdr.StreamID() + // 读取数据 + data := make([]byte, hdr.Length()) + if _, err := io.ReadFull(conn, data); err != nil { + return f, err + } + f.data = data + return f, nil +} + +// WriteFrame 将帧写入连接 +func WriteFrame(conn net.Conn, frame Frame) error { + hdr := RawHeader{} + hdr[0] = frame.ver + hdr[1] = frame.cmd + binary.LittleEndian.PutUint16(hdr[2:], uint16(len(frame.data))) + binary.LittleEndian.PutUint32(hdr[4:], uint32(frame.sid)) + + buf := hdr.Bytes() + buf = append(buf, frame.data...) + _, err := conn.Write(buf) + if err != nil { + return err + } + return nil +} + +// Version 返回协议版本 +func (f *Frame) Version() byte { + return f.ver +} + +// Cmd 返回命令字 +func (f *Frame) Cmd() byte { + return f.cmd +} + +// Length 返回数据长度 +func (f *Frame) Length() uint16 { + return uint16(len(f.data)) +} + +// Data 返回数据字节集 +func (f *Frame) Data() []byte { + return f.data +} + +// StreamID 返回流 ID +func (f *Frame) StreamID() uint32 { + return f.sid +} + +// SetData 设置数据(超出 65535 字节返回错误) +func (f *Frame) SetData(data []byte) error { + if len(data) > frameBufSize-1 { + return fmt.Errorf("超出最大长度:%d", frameBufSize-1) + } + f.data = data + return nil +} + +// Marshal 将 v 序列化为 JSON 后写入数据 +func (f *Frame) Marshal(v any) error { + buf, err := json.Marshal(v) + if err != nil { + return err + } + return f.SetData(buf) +} + +// Unmarshal 将数据按 JSON 反序列化为 v +func (f *Frame) Unmarshal(v any) error { + if f.data == nil || len(f.data) == 0 { + return errors.New("没有数据") + } + return json.Unmarshal(f.data, v) +} + +// String 返回帧的描述字符串 +func (f *Frame) String() string { + return fmt.Sprintf("Version:%d Cmd:%d StreamID:%d Length:%d", + f.Version(), f.Cmd(), f.StreamID(), f.Length()) +} + +// RawHeader 原始帧头(8 字节数组) +type RawHeader [headerSize]byte + +// Version 返回协议版本 +func (h RawHeader) Version() byte { + return h[0] +} + +// Cmd 返回命令字 +func (h RawHeader) Cmd() byte { + return h[1] +} + +// Length 返回数据长度 +func (h RawHeader) Length() uint16 { + return binary.LittleEndian.Uint16(h[2:]) +} + +// StreamID 返回流 ID +func (h RawHeader) StreamID() uint32 { + return binary.LittleEndian.Uint32(h[4:]) +} + +// Bytes 返回帧头字节集 +func (h RawHeader) Bytes() []byte { + buf := make([]byte, headerSize) + for i, _ := range buf { + buf[i] = h[i] + } + return buf +} diff --git a/frame/frame_test.go b/frame/frame_test.go new file mode 100644 index 0000000..8133783 --- /dev/null +++ b/frame/frame_test.go @@ -0,0 +1,110 @@ +package frame + +import ( + "encoding/binary" + "net" + "testing" +) + +func TestFrameRoundtrip(t *testing.T) { + c1, c2 := net.Pipe() + defer c1.Close() + defer c2.Close() + + f := NewFrame(FrameVersion, 0x10, 42) + if err := f.SetData([]byte("hello frame")); err != nil { + t.Fatalf("SetData: %v", err) + } + + go func() { + if err := WriteFrame(c1, f); err != nil { + t.Errorf("WriteFrame: %v", err) + } + }() + + got, err := ReadFrame(c2) + if err != nil { + t.Fatalf("ReadFrame: %v", err) + } + if got.Version() != FrameVersion || got.Cmd() != 0x10 || got.StreamID() != 42 { + t.Errorf("帧头不符: %s", got.String()) + } + if string(got.Data()) != "hello frame" { + t.Errorf("Data = %q, want hello frame", got.Data()) + } +} + +func TestFramePipelineRead(t *testing.T) { + // 连续两帧不丢数据(回归:旧实现用临时 bufio.Reader 会吞掉后续帧字节) + c1, c2 := net.Pipe() + defer c1.Close() + defer c2.Close() + + go func() { + a := NewFrame(1, 1, 1) + if err := a.SetData([]byte("first")); err != nil { + t.Errorf("SetData: %v", err) + } + b := NewFrame(1, 2, 2) + if err := b.SetData([]byte("second")); err != nil { + t.Errorf("SetData: %v", err) + } + if err := WriteFrame(c1, a); err != nil { + t.Errorf("WriteFrame: %v", err) + } + if err := WriteFrame(c1, b); err != nil { + t.Errorf("WriteFrame: %v", err) + } + }() + + got1, err := ReadFrame(c2) + if err != nil || string(got1.Data()) != "first" { + t.Fatalf("第一帧 = %q, %v", got1.Data(), err) + } + got2, err := ReadFrame(c2) + if err != nil || string(got2.Data()) != "second" { + t.Fatalf("第二帧 = %q, %v", got2.Data(), err) + } +} + +func TestSetDataOverflow(t *testing.T) { + f := NewFrame(1, 0, 0) + if err := f.SetData(make([]byte, 65536)); err == nil { + t.Error("超过 65535 字节应返回错误") + } + if err := f.SetData(make([]byte, 65535)); err != nil { + t.Errorf("65535 字节应成功: %v", err) + } +} + +func TestMarshalUnmarshal(t *testing.T) { + f := NewFrame(1, 3, 9) + type payload struct { + Name string + N int + } + if err := f.Marshal(payload{Name: "a", N: 1}); err != nil { + t.Fatalf("Marshal: %v", err) + } + var got payload + if err := f.Unmarshal(&got); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if got.Name != "a" || got.N != 1 { + t.Errorf("Unmarshal 结果 = %+v", got) + } +} + +func TestRawHeader(t *testing.T) { + h := RawHeader{} + h[0] = 2 + h[1] = 7 + binary.LittleEndian.PutUint16(h[2:], 300) + binary.LittleEndian.PutUint32(h[4:], 100000) + if h.Version() != 2 || h.Cmd() != 7 || h.Length() != 300 || h.StreamID() != 100000 { + t.Errorf("RawHeader 解析不符: %v %v %v %v", h.Version(), h.Cmd(), h.Length(), h.StreamID()) + } + if len(h.Bytes()) != 8 { + t.Errorf("Bytes 长度 = %d, want 8", len(h.Bytes())) + } +}