feat(frame): go-hua/common 帧协议独立成包并修复两处缺陷
- SetData 超长时 fmt.Errorf 结果未 return,超 65535 字节静默放行 → 改为返回错误 - ReadFrame 原实现 conn.Read 直读帧头(短读错位)且一次性 bufio.Reader 会丢弃 缓冲中的后续帧字节 → 统一改 io.ReadFull 精确读取,新增连续两帧回归测试 - 补齐单元测试(往返/流水线/超长/Marshal/RawHeader)、examples/frame 例程与 README
This commit is contained in:
@@ -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)
|
||||||
|
}
|
||||||
@@ -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 持有。
|
||||||
+172
@@ -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
|
||||||
|
}
|
||||||
@@ -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()))
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user