handle sticky packet (#39)

close #38

Author: chenjunbiao

README_CN.md

@@ -414,16 +414,19 @@ More examples in `/pkg/spec/encoding/pvarint_test.go|varfloat_test.go`
 - [x] v0.4.0 - 支持[yomo-thermometer-plugin](https://github.com/10cella/yomo-thermometer-plugin)插件
   - [x] 支持[]Thermometer{}的Mold形式
   - [x] [YoMo](https://github.com/yomorun/yomo)框架的正式切换至[Yomo-codec-golang](https://github.com/yomorun/yomo-codec-golang)
-- [x] v0.5.0 - 性能压测及优化重构
-- [ ] v0.6.0 - 支持UUID及Map类型
+- [x] v1.0.0 - 性能压测及优化重构
+  - [x] YomoCodec的性能测试，与JSON实现版本的对比
+  - [x] 通过跳KEY解码字节流优化性能
+  - [x] 粘包处理
+- [ ] v1.1.0 - 支持UUID及Map类型
   - [ ] encoding
   - [ ] PrimitivePacket
   - [ ] ProtoCode
-- [ ] v0.7.0 - 支持[YoMo](https://github.com/yomorun/yomo)框架的新的解析需求
-  - [ ] 满足处理流程：解析--监听--读取--处理--写入
-  - [ ] 支持两种出理流程的切换
-- [ ] v0.9.0 - 支持简单的控制指令和错误码
-- [ ] v1.0.0 - 支持与QUIC-for-YoMo框架的集成
+- [ ] v1.2.0 - 支持[YoMo](https://github.com/yomorun/yomo)框架的新的解析需求
+  - [ ] 满足处理流程
+  - [ ] 支持两种处理流程的切换
+- [ ] v1.3.0 - 支持简单的控制指令和错误码
+- [ ] v2.0.0 - 支持与Quicy框架的集成
 
 ## YoMo Codec
 

docs/report1.jpg

@@ -36,3 +36,7 @@ func NewTag(b byte) *Tag {
 func (t *Tag) IsArray() bool {
 	return t.raw&utils.ArrayFlag == utils.ArrayFlag
 }
+
+func (t *Tag) Raw() byte {
+	return t.raw
+}

docs/report2.jpg

@@ -11,7 +11,6 @@ import (
 
 // collectingCodec: Implementation of the YomoCodec Interface
 type collectingCodec struct {
-	//Value []byte
 	Value *y3.NodePacket
 
 	Tag *ycodec.Tag
@@ -20,42 +19,34 @@ type collectingCodec struct {
 	Length    int32
 	Size      int32
 
-	//Observe string
 	Observe byte
 	Sbuf    []byte
 
-	//Result       [][]byte
 	Result       []*y3.NodePacket
 	OriginResult [][]byte
 
-	//ProtoCodec
 	proto ProtoCodec
 }
 
 func NewCollectingCodec(observe string) YomoCodec {
 	ob := packetutils.KeyOf(observe)
 	codec := &collectingCodec{
-		//Value:     make([]byte, 0),
-		Value:     nil,
-		Tag:       nil,
-		LengthBuf: make([]byte, 0),
-		Length:    0,
-		Size:      0,
-		Sbuf:      make([]byte, 0),
-		Observe:   ob,
-		//Result:       make([][]byte, 0),
+		Value:        nil,
+		Tag:          nil,
+		LengthBuf:    make([]byte, 0),
+		Length:       0,
+		Size:         0,
+		Sbuf:         make([]byte, 0),
+		Observe:      ob,
 		Result:       make([]*y3.NodePacket, 0),
 		OriginResult: make([][]byte, 0),
-		//ProtoCodec
-		proto: NewProtoCodec(ob),
+		proto:        NewProtoCodec(ob),
 	}
 
 	return codec
 }
 
-// #2 [优化] 跳key优化
 func (codec *collectingCodec) Decoder(buf []byte) {
-	//key := packetutils.KeyOf(codec.Observe)
 	key := codec.Observe
 	for _, c := range buf {
 		// tag
@@ -80,33 +71,26 @@ func (codec *collectingCodec) Decoder(buf []byte) {
 
 		codec.Sbuf = append(codec.Sbuf, c)
 
-		// buf end, then handle Sbuf
 		if int32(len(codec.Sbuf)) == 1+codec.Size+codec.Length {
-			// Decode Packet from Sbuf
 			packet, _, err := y3.DecodeNodePacket(codec.Sbuf)
 			if err != nil {
 				logger.Errorf("::Decoder DecodeNodePacket error:%v", err)
 				codec.reset()
 				continue
 			}
 
-			// temp save Sbuf and reset
-			//result := make([]byte, 0)
 			var result *y3.NodePacket
 			originResult := codec.Sbuf
 			codec.reset()
 
 			//matching
 			var matching = false
 			flag, _, _ := packetutils.MatchingKey(key, packet)
-			//if flag || []byte("*")[0] == key {
 			if flag {
 				matching = true
-				//result = originResult
 				result = packet
 			}
 
-			// save to result
 			if matching {
 				codec.Result = append(codec.Result, result)
 				codec.OriginResult = append(codec.OriginResult, placeholder)
@@ -144,16 +128,13 @@ func (codec *collectingCodec) Read(mold interface{}) (interface{}, error) {
 	result := codec.Result[0]
 	codec.Result = codec.Result[1:]
 
-	//proto := NewProtoCodec(codec.Observe)
 	proto := codec.proto
 	if proto.IsStruct(mold) {
-		//err := proto.UnmarshalStruct(result, mold)
 		err := proto.UnmarshalStructByNodePacket(result, mold)
 		if err != nil {
 			return nil, err
 		}
 	} else {
-		//err := proto.UnmarshalBasic(result, &mold)
 		err := proto.UnmarshalBasicByNodePacket(result, &mold)
 		if err != nil {
 			return nil, err
@@ -168,11 +149,6 @@ func (codec *collectingCodec) Read(mold interface{}) (interface{}, error) {
 
 // Write: write interface to stream
 func (codec *collectingCodec) Write(w io.Writer, T interface{}, mold interface{}) (int, error) {
-	// #1. mold --> NodePacket
-	// #2. merge NodePacket --> codec.Value NodePacket
-	// #3. NodePacket --> []byte
-	// #4. Write []byte
-	//proto := NewProtoCodec(codec.Observe)
 	proto := codec.proto
 	buf, err := proto.MarshalNative(T)
 	if err != nil {
@@ -195,17 +171,14 @@ func (codec *collectingCodec) Encoder(buf []byte, mold interface{}) ([]byte, err
 	for _, data := range codec.OriginResult {
 		index = index + 1
 		if codec.isDecoder(data) {
-			//source, _, _ := y3.DecodeNodePacket(codec.Value)
 			source := codec.Value
 
-			//key := packetutils.KeyOf(codec.Observe)
 			key := codec.Observe
 			_buf, err := codec.mergePacket(source, key, buf, mold)
 			if err != nil {
 				return nil, err
 			}
 
-			//codec.Value = make([]byte, 0)
 			codec.Value = nil
 			result = append(result, _buf...)
 			break