feat: chunk add end reason field

Author: tangtaoit

chunk.go

@@ -1,11 +1,49 @@
 package wkproto
 
+type EndReason uint8
+
+// EndReason constants define why a stream was completed
+const (
+	// EndReasonSuccess indicates the stream completed successfully (default)
+	EndReasonSuccess EndReason = 0
+	// EndReasonTimeout indicates the stream ended due to inactivity timeout
+	EndReasonTimeout EndReason = 1
+	// EndReasonError indicates the stream ended due to an error
+	EndReasonError EndReason = 2
+	// EndReasonCancelled indicates the stream was manually cancelled
+	EndReasonCancelled EndReason = 3
+	// EndReasonForce indicates the stream was forcefully ended (e.g., channel closure)
+	EndReasonForce EndReason = 4
+)
+
+func (e EndReason) String() string {
+	switch e {
+	case EndReasonSuccess:
+		return "success"
+	case EndReasonTimeout:
+		return "timeout"
+	case EndReasonError:
+		return "error"
+	case EndReasonCancelled:
+		return "cancelled"
+	case EndReasonForce:
+		return "force"
+	default:
+		return "unknown"
+	}
+}
+
+func (e EndReason) Value() uint8 {
+	return uint8(e)
+}
+
 // ChunkPacket 消息块
 type ChunkPacket struct {
 	Framer
-	MessageID int64  // 消息ID(同个消息多个块的消息ID相同)
-	ChunkID   uint64 // 块ID（顺序递增）
-	Payload   []byte // 消息内容
+	MessageID int64     // 消息ID(同个消息多个块的消息ID相同)
+	ChunkID   uint64    // 块ID（顺序递增）
+	EndReason EndReason // 结束原因
+	Payload   []byte    // 消息内容
 }
 
 // GetPacketType 获得包类型
@@ -24,6 +62,7 @@ func (c *ChunkPacket) SizeWithProtoVersion(protVersion uint8) int {
 func encodeChunk(chunkPacket *ChunkPacket, enc *Encoder, _ uint8) error {
 	enc.WriteInt64(chunkPacket.MessageID)
 	enc.WriteUint64(chunkPacket.ChunkID)
+	enc.WriteUint8(chunkPacket.EndReason.Value())
 	enc.WriteBytes(chunkPacket.Payload)
 	return nil
 }
@@ -41,6 +80,12 @@ func decodeChunk(frame Frame, data []byte, _ uint8) (Frame, error) {
 		return nil, err
 	}
 
+	var endReason uint8
+	if endReason, err = dec.Uint8(); err != nil {
+		return nil, err
+	}
+	chunkPacket.EndReason = EndReason(endReason)
+
 	if chunkPacket.Payload, err = dec.BinaryAll(); err != nil {
 		return nil, err
 	}
@@ -53,6 +98,7 @@ func encodeChunkSize(packet *ChunkPacket, _ uint8) int {
 	size := 0
 	size += MessageIDByteSize   // 消息ID
 	size += ChunkIDByteSize     // 块ID
+	size += EndReasonByteSize   // 结束原因
 	size += len(packet.Payload) // 消息内容
 	return size
 }

chunk_test.go

@@ -11,6 +11,7 @@ func TestChunkEncodeAndDecode(t *testing.T) {
 		MessageID: 1,
 		ChunkID:   1,
 		Payload:   []byte("hello world"),
+		EndReason: 1,
 	}
 
 	codec := New()

common.go

@@ -13,12 +13,16 @@ type Framer struct {
 	SyncOnce         bool   // 此消息只被同步或被消费一次
 	DUP              bool   // 是否是重发消息
 	HasServerVersion bool   // 是否有服务端版本 connack包用到
-
-	FrameSize int64
+	End              bool   // 消息块：是否是最后一包
+	FrameSize        int64
 }
 
 // ToFixHeaderUint8 ToFixHeaderUint8
 func ToFixHeaderUint8(f Frame) uint8 {
+
+	if f.GetFrameType() == Chunk {
+		return byte(int(f.GetFrameType()<<4) | encodeBool(f.GetEnd()))
+	}
 	typeAndFlags := encodeBool(f.GetDUP())<<3 | encodeBool(f.GetsyncOnce())<<2 | encodeBool(f.GetRedDot())<<1 | encodeBool(f.GetNoPersist())
 	if f.GetFrameType() == CONNACK {
 		typeAndFlags = encodeBool(f.GetHasServerVersion())
@@ -34,8 +38,12 @@ func FramerFromUint8(v uint8) Framer {
 	p.SyncOnce = (v >> 2 & 0x01) > 0
 	p.DUP = (v >> 3 & 0x01) > 0
 	p.FrameType = FrameType(v >> 4)
-	if p.FrameType == CONNACK {
+
+	switch p.FrameType {
+	case CONNACK:
 		p.HasServerVersion = (v & 0x01) > 0
+	case Chunk:
+		p.End = (v & 0x01) > 0
 	}
 
 	return p
@@ -79,6 +87,10 @@ func (f Framer) GetHasServerVersion() bool {
 	return f.HasServerVersion
 }
 
+func (f Framer) GetEnd() bool {
+	return f.End
+}
+
 func (f Framer) String() string {
 	return fmt.Sprintf("packetType: %s remainingLength:%d NoPersist:%v redDot:%v syncOnce:%v DUP:%v", f.GetFrameType().String(), f.RemainingLength, f.NoPersist, f.RedDot, f.SyncOnce, f.DUP)
 }
@@ -327,6 +339,8 @@ type Frame interface {
 	GetDUP() bool
 	GetFrameSize() int64       // 总个frame的大小（不参与编码解码）
 	GetHasServerVersion() bool // 是否有服务端版本 connack包用到
+
+	GetEnd() bool
 }
 
 type Channel struct {
@@ -353,6 +367,7 @@ const (
 	ExpireByteSize          = 4
 	NodeIdByteSize          = 8
 	ChunkIDByteSize         = 8
+	EndReasonByteSize       = 1
 )
 
 const (