+// {@link LengthFieldBasedFrameDecoder} has many configuration parameters so +// that it can decode any message with a length field, which is often seen in +// proprietary client-server protocols. Here are some example that will give +// you the basic idea on which option does what. +// +//
+// LengthFieldOffset = 0 +// LengthFieldLength = 2 +// LengthAdjustment = 0 +// InitialBytesToStrip = 0 (= do not strip header) +// +// BEFORE DECODE (14 bytes) AFTER DECODE (14 bytes) +// +--------+----------------+ +--------+----------------+ +// | Length | Actual Content |----->| Length | Actual Content | +// | 0x000C | "HELLO, WORLD" | | 0x000C | "HELLO, WORLD" | +// +--------+----------------+ +--------+----------------+ +//+// +//
+// LengthFieldOffset = 0 +// LengthFieldLength = 2 +// LengthAdjustment = 0 +// InitialBytesToStrip = 2 (= the length of the Length field) +// +// BEFORE DECODE (14 bytes) AFTER DECODE (12 bytes) +// +--------+----------------+ +----------------+ +// | Length | Actual Content |----->| Actual Content | +// | 0x000C | "HELLO, WORLD" | | "HELLO, WORLD" | +// +--------+----------------+ +----------------+ +//+// +//
+// LengthFieldOffset = 0 +// LengthFieldLength = 2 +// LengthAdjustment = -2 (= the length of the Length field) +// InitialBytesToStrip = 0 +// +// BEFORE DECODE (14 bytes) AFTER DECODE (14 bytes) +// +--------+----------------+ +--------+----------------+ +// | Length | Actual Content |----->| Length | Actual Content | +// | 0x000E | "HELLO, WORLD" | | 0x000E | "HELLO, WORLD" | +// +--------+----------------+ +--------+----------------+ +//+// +//
+// LengthFieldOffset = 2 (= the length of Header 1) +// LengthFieldLength = 3 +// LengthAdjustment = 0 +// InitialBytesToStrip = 0 +// +// BEFORE DECODE (17 bytes) AFTER DECODE (17 bytes) +// +----------+----------+----------------+ +----------+----------+----------------+ +// | Header 1 | Length | Actual Content |----->| Header 1 | Length | Actual Content | +// | 0xCAFE | 0x00000C | "HELLO, WORLD" | | 0xCAFE | 0x00000C | "HELLO, WORLD" | +// +----------+----------+----------------+ +----------+----------+----------------+ +//+// +//
+// LengthFieldOffset = 0 +// LengthFieldLength = 3 +// LengthAdjustment = 2 (= the length of Header 1) +// InitialBytesToStrip = 0 +// +// BEFORE DECODE (17 bytes) AFTER DECODE (17 bytes) +// +----------+----------+----------------+ +----------+----------+----------------+ +// | Length | Header 1 | Actual Content |----->| Length | Header 1 | Actual Content | +// | 0x00000C | 0xCAFE | "HELLO, WORLD" | | 0x00000C | 0xCAFE | "HELLO, WORLD" | +// +----------+----------+----------------+ +----------+----------+----------------+ +//+// +//
+// LengthFieldOffset = 1 (= the length of HDR1) +// LengthFieldLength = 2 +// LengthAdjustment = 1 (= the length of HDR2) +// InitialBytesToStrip = 3 (= the length of HDR1 + LEN) +// +// BEFORE DECODE (16 bytes) AFTER DECODE (13 bytes) +// +------+--------+------+----------------+ +------+----------------+ +// | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content | +// | 0xCA | 0x000C | 0xFE | "HELLO, WORLD" | | 0xFE | "HELLO, WORLD" | +// +------+--------+------+----------------+ +------+----------------+ +//+// +//
+// LengthFieldOffset = 1
+// LengthFieldLength = 2
+// LengthAdjustment = -3 (= the length of HDR1 + LEN, negative)
+// InitialBytesToStrip = 3
+//
+// BEFORE DECODE (16 bytes) AFTER DECODE (13 bytes)
+// +------+--------+------+----------------+ +------+----------------+
+// | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
+// | 0xCA | 0x0010 | 0xFE | "HELLO, WORLD" | | 0xFE | "HELLO, WORLD" |
+// +------+--------+------+----------------+ +------+----------------+
+// https://blog.csdn.net/weixin_45271492/article/details/125347939
+
+type EncoderData struct {
+ lengthField ziface.LengthField
+ LengthFieldEndOffset int //长度字段结束位置的偏移量 LengthFieldOffset+LengthFieldLength
+ failFast bool //快速失败
+ discardingTooLongFrame bool //true 表示开启丢弃模式，false 正常工作模式
+ tooLongFrameLength int64 //当某个数据包的长度超过maxLength，则开启丢弃模式，此字段记录需要丢弃的数据长度
+ bytesToDiscard int64 //记录还剩余多少字节需要丢弃
+ in []byte
+ lock sync.Mutex
+}
+
+func NewLengthFieldFrameDecoderByLengthField(lengthField ziface.LengthField) ziface.IDecoder {
+ return &EncoderData{
+ lengthField: lengthField,
+ LengthFieldEndOffset: lengthField.LengthFieldOffset + lengthField.LengthFieldLength,
+ in: make([]byte, 0),
+ }
+}
+
+func NewLengthFieldFrameDecoder(maxFrameLength int64, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip int) ziface.IDecoder {
+ return &EncoderData{
+ lengthField: ziface.LengthField{
+ MaxFrameLength: maxFrameLength,
+ LengthFieldOffset: lengthFieldOffset,
+ LengthFieldLength: lengthFieldLength,
+ LengthAdjustment: lengthAdjustment,
+ InitialBytesToStrip: initialBytesToStrip,
+ Order: binary.BigEndian,
+ },
+ LengthFieldEndOffset: lengthFieldOffset + lengthFieldLength,
+ in: make([]byte, 0),
+ }
+}
+
+func (this *EncoderData) fail(frameLength int64) {
+ //丢弃完成或未完成都抛异常
+ //if frameLength > 0 {
+ // msg := fmt.Sprintf("Adjusted frame length exceeds %d : %d - discarded", this.MaxFrameLength, frameLength)
+ // panic(msg)
+ //} else {
+ // msg := fmt.Sprintf("Adjusted frame length exceeds %d - discarded", this.MaxFrameLength)
+ // panic(msg)
+ //}
+}
+
+func (this *EncoderData) discardingTooLongFrameFunc(buffer *bytes.Buffer) {
+ //保存还需丢弃多少字节
+ bytesToDiscard := this.bytesToDiscard
+ //获取当前可以丢弃的字节数，有可能出现半包
+ localBytesToDiscard := math.Min(float64(bytesToDiscard), float64(buffer.Len()))
+ //fmt.Println("--->", bytesToDiscard, buffer.Len(), localBytesToDiscard)
+ localBytesToDiscard = 2
+ //丢弃
+ buffer.Next(int(localBytesToDiscard))
+ //更新还需丢弃的字节数
+ bytesToDiscard -= int64(localBytesToDiscard)
+ this.bytesToDiscard = bytesToDiscard
+ //是否需要快速失败，回到上面的逻辑
+ this.failIfNecessary(false)
+}
+
+func (this *EncoderData) getUnadjustedFrameLength(buf *bytes.Buffer, offset int, length int, order binary.ByteOrder) int64 {
+ //长度字段的值
+ var frameLength int64
+ arr := buf.Bytes()
+ arr = arr[offset : offset+length]
+ buffer := bytes.NewBuffer(arr)
+ switch length {
+ case 1:
+ //byte
+ var value byte
+ binary.Read(buffer, order, &value)
+ frameLength = int64(value)
+ case 2:
+ //short
+ var value int16
+ binary.Read(buffer, order, &value)
+ frameLength = int64(value)
+ case 3:
+ //int占32位，这里取出后24位，返回int类型
+ if order == binary.LittleEndian {
+ n := int(uint(arr[0]) | uint(arr[1])<<8 | uint(arr[2])<<16)
+ frameLength = int64(n)
+ } else {
+ n := int(uint(arr[2]) | uint(arr[1])<<8 | uint(arr[0])<<16)
+ frameLength = int64(n)
+ }
+ case 4:
+ //int
+ var value int32
+ binary.Read(buffer, order, &value)
+ frameLength = int64(value)
+ case 8:
+ //long
+ binary.Read(buffer, order, &frameLength)
+ default:
+ panic(fmt.Sprintf("unsupported LengthFieldLength: %d (expected: 1, 2, 3, 4, or 8)", this.lengthField.LengthFieldLength))
+ }
+ return frameLength
+}
+
+func (this *EncoderData) failOnNegativeLengthField(in *bytes.Buffer, frameLength int64, lengthFieldEndOffset int) {
+ in.Next(lengthFieldEndOffset)
+ panic(fmt.Sprintf("negative pre-adjustment length field: %d", frameLength))
+}
+
+func (this *EncoderData) failIfNecessary(firstDetectionOfTooLongFrame bool) {
+ if this.bytesToDiscard == 0 {
+ //说明需要丢弃的数据已经丢弃完成
+ //保存一下被丢弃的数据包长度
+ tooLongFrameLength := this.tooLongFrameLength
+ this.tooLongFrameLength = 0
+ //关闭丢弃模式
+ this.discardingTooLongFrame = false
+ //failFast：默认true
+ //firstDetectionOfTooLongFrame：传入true
+ if !this.failFast || firstDetectionOfTooLongFrame {
+ //快速失败
+ this.fail(tooLongFrameLength)
+ }
+ } else {
+ //说明还未丢弃完成
+ if this.failFast && firstDetectionOfTooLongFrame {
+ //快速失败
+ this.fail(this.tooLongFrameLength)
+ }
+ }
+}
+
+// frameLength：数据包的长度
+func (this *EncoderData) exceededFrameLength(in *bytes.Buffer, frameLength int64) {
+ //数据包长度-可读的字节数 两种情况
+ //1. 数据包总长度为100，可读的字节数为50，说明还剩余50个字节需要丢弃但还未接收到
+ //2. 数据包总长度为100，可读的字节数为150，说明缓冲区已经包含了整个数据包
+ discard := frameLength - int64(in.Len())
+ //记录一下最大的数据包的长度
+ this.tooLongFrameLength = frameLength
+ if discard < 0 {
+ //说明是第二种情况，直接丢弃当前数据包
+ in.Next(int(frameLength))
+ } else {
+ //说明是第一种情况，还有部分数据未接收到
+ //开启丢弃模式
+ this.discardingTooLongFrame = true
+ //记录下次还需丢弃多少字节
+ this.bytesToDiscard = discard
+ //丢弃缓冲区所有数据
+ in.Next(in.Len())
+ }
+ //跟进去
+ this.failIfNecessary(true)
+}
+
+func (this *EncoderData) failOnFrameLengthLessThanInitialBytesToStrip(in *bytes.Buffer, frameLength int64, initialBytesToStrip int) {
+ in.Next(int(frameLength))
+ panic(fmt.Sprintf("Adjusted frame length (%d) is less than InitialBytesToStrip: %d", frameLength, initialBytesToStrip))
+}
+
+// https://blog.csdn.net/qq_39280718/article/details/125762004
+func (this *EncoderData) decode(buf []byte) []byte {
+ in := bytes.NewBuffer(buf)
+ //丢弃模式
+ if this.discardingTooLongFrame {
+ this.discardingTooLongFrameFunc(in)
+ }
+ ////判断缓冲区中可读的字节数是否小于长度字段的偏移量
+ if in.Len() < this.LengthFieldEndOffset {
+ //说明长度字段的包都还不完整，半包
+ return nil
+ }
+ //执行到这，说明可以解析出长度字段的值了
+
+ //计算出长度字段的开始偏移量
+ actualLengthFieldOffset := this.lengthField.LengthFieldOffset
+ //获取长度字段的值，不包括lengthAdjustment的调整值
+ frameLength := this.getUnadjustedFrameLength(in, actualLengthFieldOffset, this.lengthField.LengthFieldLength, this.lengthField.Order)
+ //如果数据帧长度小于0，说明是个错误的数据包
+ if frameLength < 0 {
+ //内部会跳过这个数据包的字节数，并抛异常
+ this.failOnNegativeLengthField(in, frameLength, this.LengthFieldEndOffset)
+ }
+
+ //套用前面的公式：长度字段后的数据字节数=长度字段的值+lengthAdjustment
+ //frameLength就是长度字段的值，加上lengthAdjustment等于长度字段后的数据字节数
+ //lengthFieldEndOffset为lengthFieldOffset+lengthFieldLength
+ //那说明最后计算出的framLength就是整个数据包的长度
+ frameLength += int64(this.lengthField.LengthAdjustment) + int64(this.LengthFieldEndOffset)
+ //丢弃模式就是在这开启的
+ //如果数据包长度大于最大长度
+ if frameLength > int64(this.lengthField.MaxFrameLength) {
+ //对超过的部分进行处理
+ this.exceededFrameLength(in, frameLength)
+ return nil
+ }
+
+ //执行到这说明是正常模式
+ //数据包的大小
+ frameLengthInt := int(frameLength)
+ //判断缓冲区可读字节数是否小于数据包的字节数
+ if in.Len() < frameLengthInt {
+ //半包，等会再来解析
+ return nil
+ }
+
+ //执行到这说明缓冲区的数据已经包含了数据包
+
+ //跳过的字节数是否大于数据包长度
+ if this.lengthField.InitialBytesToStrip > frameLengthInt {
+ this.failOnFrameLengthLessThanInitialBytesToStrip(in, frameLength, this.lengthField.InitialBytesToStrip)
+ }
+ //跳过initialBytesToStrip个字节
+ in.Next(this.lengthField.InitialBytesToStrip)
+ //解码
+ //获取跳过后的真实数据长度
+ actualFrameLength := frameLengthInt - this.lengthField.InitialBytesToStrip
+ //提取真实的数据
+ buff := make([]byte, actualFrameLength)
+ in.Read(buff)
+ //bytes.NewBuffer([]byte{})
+ //_in := bytes.NewBuffer(buff)
+ return buff
+}
+
+func (this *EncoderData) Decode(buff []byte) [][]byte {
+ this.lock.Lock()
+ defer this.lock.Unlock()
+ this.in = append(this.in, buff...)
+ resp := make([][]byte, 0)
+ for {
+ arr := this.decode(this.in)
+ if arr != nil {
+ //证明已经解析出一个完整包
+ resp = append(resp, arr)
+ _size := len(arr)
+ //_len := len(this.in)
+ //fmt.Println(_len)
+ if _size > 0 {
+ this.in = this.in[_size:]
+ }
+ } else {
+ return resp
+ }
+ }
+ return nil
+
+}
diff --git a/zcode/lengthfieldframeinterceptor.go b/zcode/lengthfieldframeinterceptor.go
new file mode 100644
index 00000000..0ff0510c
--- /dev/null
+++ b/zcode/lengthfieldframeinterceptor.go
@@ -0,0 +1,54 @@
+/**
+ * @author uuxia
+ * @date 15:58 2023/3/10
+ * @description 通过拦截，处理数据，任务向下传递
+ **/
+
+package zcode
+
+import (
+ "github.com/aceld/zinx/ziface"
+)
+
+type LengthFieldFrameInterceptor struct {
+ decoder ziface.IDecoder
+}
+
+func NewLengthFieldFrameInterceptor(maxFrameLength int64, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip int) *LengthFieldFrameInterceptor {
+ return &LengthFieldFrameInterceptor{
+ decoder: NewLengthFieldFrameDecoder(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip),
+ }
+}
+
+func (this *LengthFieldFrameInterceptor) Intercept(chain ziface.Chain) ziface.Response {
+ request := chain.Request()
+ if request != nil {
+ switch request.(type) {
+ case ziface.IRequest:
+ iRequest := request.(ziface.IRequest)
+ iMessage := iRequest.GetMessage()
+ if iMessage != nil {
+ data := iMessage.GetData()
+ if this.decoder != nil {
+ bytebuffers := this.decoder.Decode(data)
+ size := len(bytebuffers)
+ if size == 0 { //半包，或者其他情况，任务就不要往下再传递了
+ return nil
+ }
+ for i := 0; i < size; i++ {
+ buffer := bytebuffers[i]
+ if buffer != nil {
+ bufferSize := len(buffer)
+ iMessage.SetData(buffer)
+ iMessage.SetDataLen(uint32(bufferSize))
+ if i < size-1 {
+ chain.Proceed(chain.Request())
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ return chain.Proceed(chain.Request())
+}
diff --git a/zcode/realinterceptorchain.go b/zcode/realinterceptorchain.go
new file mode 100644
index 00000000..d3ed9fe5
--- /dev/null
+++ b/zcode/realinterceptorchain.go
@@ -0,0 +1,37 @@
+/**
+ * @author uuxia
+ * @date 15:56 2023/3/10
+ * @description 责任链模式
+ **/
+
+package zcode
+
+import "github.com/aceld/zinx/ziface"
+
+type RealInterceptorChain struct {
+ request ziface.Request
+ position int
+ interceptors []ziface.Interceptor
+}
+
+func (this *RealInterceptorChain) Request() ziface.Request {
+ return this.request
+}
+
+func (this *RealInterceptorChain) Proceed(request ziface.Request) ziface.Response {
+ if this.position < len(this.interceptors) {
+ chain := NewRealInterceptorChain(this.interceptors, this.position+1, request)
+ interceptor := this.interceptors[this.position]
+ response := interceptor.Intercept(chain)
+ return response
+ }
+ return request
+}
+
+func NewRealInterceptorChain(list []ziface.Interceptor, pos int, request ziface.Request) ziface.Chain {
+ return &RealInterceptorChain{
+ request: request,
+ position: pos,
+ interceptors: list,
+ }
+}
diff --git a/ziface/iDecoder.go b/ziface/iDecoder.go
new file mode 100644
index 00000000..78b059d9
--- /dev/null
+++ b/ziface/iDecoder.go
@@ -0,0 +1,20 @@
+package ziface
+
+import "encoding/binary"
+
+type IDecoder interface {
+ Decode(buff []byte) [][]byte
+}
+
+type LengthField struct {
+ //大小端排序
+ //大端模式：是指数据的高字节保存在内存的低地址中，而数据的低字节保存在内存的高地址中，地址由小向大增加，而数据从高位往低位放；
+ //小端模式：是指数据的高字节保存在内存的高地址中，而数据的低字节保存在内存的低地址中，高地址部分权值高，低地址部分权值低，和我们的日常逻辑方法一致。
+ //不了解的自行查阅一下资料
+ Order binary.ByteOrder
+ MaxFrameLength int64 //最大帧长度
+ LengthFieldOffset int //长度字段偏移量
+ LengthFieldLength int //长度域字段的字节数
+ LengthAdjustment int //长度调整
+ InitialBytesToStrip int //需要跳过的字节数
+}
diff --git a/ziface/iclient.go b/ziface/iclient.go
index 38a9fac7..13ab5438 100644
--- a/ziface/iclient.go
+++ b/ziface/iclient.go
@@ -1,12 +1,13 @@
// Package ziface 主要提供zinx全部抽象层接口定义.
// 包括:
-// IServer 服务mod接口
-// IRouter 路由mod接口
-// IConnection 连接mod层接口
-// IMessage 消息mod接口
-// IDataPack 消息拆解接口
-// IMsgHandler 消息处理及协程池接口
-// IClient 客户端接口
+//
+// IServer 服务mod接口
+// IRouter 路由mod接口
+// IConnection 连接mod层接口
+// IMessage 消息mod接口
+// IDataPack 消息拆解接口
+// IMsgHandler 消息处理及协程池接口
+// IClient 客户端接口
//
// 当前文件描述:
// @Title iclient.go
@@ -30,4 +31,6 @@ type IClient interface {
GetMsgHandler() IMsgHandle //获取Client绑定的消息处理模块
StartHeartBeat(time.Duration) //启动心跳检测
StartHeartBeatWithOption(time.Duration, *HeartBeatOption) //启动心跳检测(自定义回调)
+ AddInterceptor(interceptor Interceptor) //添加协议解析拦截器
+ GetLengthField() LengthField
}
diff --git a/ziface/iconnection.go b/ziface/iconnection.go
index 88592963..b69dc948 100644
--- a/ziface/iconnection.go
+++ b/ziface/iconnection.go
@@ -1,11 +1,12 @@
// Package ziface 主要提供zinx全部抽象层接口定义.
// 包括:
-// IServer 服务mod接口
-// IRouter 路由mod接口
-// IConnection 连接mod层接口
-// IMessage 消息mod接口
-// IDataPack 消息拆解接口
-// IMsgHandler 消息处理及协程池接口
+//
+// IServer 服务mod接口
+// IRouter 路由mod接口
+// IConnection 连接mod层接口
+// IMessage 消息mod接口
+// IDataPack 消息拆解接口
+// IMsgHandler 消息处理及协程池接口
//
// 当前文件描述:
// @Title iconnection.go
@@ -18,7 +19,7 @@ import (
"net"
)
-//定义连接接口
+// 定义连接接口
type IConnection interface {
Start() //启动连接，让当前连接开始工作
Stop() //停止连接，结束当前连接状态
@@ -29,6 +30,8 @@ type IConnection interface {
RemoteAddr() net.Addr //获取链接远程地址信息
LocalAddr() net.Addr //获取链接本地地址信息
+ Send(data []byte) error
+ SendToQueue(data []byte) error
SendMsg(msgID uint32, data []byte) error //直接将Message数据发送数据给远程的TCP客户端(无缓冲)
SendBuffMsg(msgID uint32, data []byte) error //直接将Message数据发送给远程的TCP客户端(有缓冲)
@@ -38,10 +41,10 @@ type IConnection interface {
IsAlive() bool //判断当前连接是否存活
}
-//用户自定义的心跳检测消息处理方法
+// 用户自定义的心跳检测消息处理方法
type HeartBeatMsgFunc func(IConnection) []byte
-//用户自定义的远程连接不存活时的处理方法
+// 用户自定义的远程连接不存活时的处理方法
type OnRemoteNotAlive func(IConnection)
type HeartBeatOption struct {
diff --git a/ziface/imsghandler.go b/ziface/imsghandler.go
index f9a30406..887d15be 100644
--- a/ziface/imsghandler.go
+++ b/ziface/imsghandler.go
@@ -1,11 +1,12 @@
// Package ziface 主要提供zinx全部抽象层接口定义.
// 包括:
-// IServer 服务mod接口
-// IRouter 路由mod接口
-// IConnection 连接mod层接口
-// IMessage 消息mod接口
-// IDataPack 消息拆解接口
-// IMsgHandler 消息处理及协程池接口
+//
+// IServer 服务mod接口
+// IRouter 路由mod接口
+// IConnection 连接mod层接口
+// IMessage 消息mod接口
+// IDataPack 消息拆解接口
+// IMsgHandler 消息处理及协程池接口
//
// 当前文件描述:
// @Title imsghandler.go
@@ -14,7 +15,7 @@
package ziface
/*
- 消息管理抽象层
+消息管理抽象层
*/
type IMsgHandle interface {
DoMsgHandler(request IRequest) //马上以非阻塞方式处理消息
@@ -22,4 +23,7 @@ type IMsgHandle interface {
AddRouter(msgID uint32, router IRouter)
StartWorkerPool() //启动worker工作池
SendMsgToTaskQueue(request IRequest) //将消息交给TaskQueue,由worker进行处理
+
+ Decode(request IRequest) //
+ AddInterceptor(interceptor Interceptor) //注册责任链任务入口，每个拦截器处理完后，数据都会传递至下一个拦截器，使得消息可以层层处理层层传递，顺序取决于注册顺序
}
diff --git a/ziface/interceptor.go b/ziface/interceptor.go
new file mode 100644
index 00000000..d9c715ea
--- /dev/null
+++ b/ziface/interceptor.go
@@ -0,0 +1,28 @@
+/**
+ * @author uuxia
+ * @date 15:54 2023/3/10
+ * @description //TODO
+ **/
+
+package ziface
+
+// 请求父类，定义空接口，用于扩展支持任意类型
+
+type Request interface {
+}
+
+// 回复父类，定义空接口，用于扩展支持任意类型
+
+type Response interface {
+}
+type Interceptor interface {
+ Intercept(Chain) Response
+}
+type Chain interface {
+ Request() Request
+ Proceed(Request) Response
+}
+type InterceptorBuilder interface {
+ AddInterceptor(interceptor Interceptor)
+ Execute(request Request) Response
+}
diff --git a/ziface/irequest.go b/ziface/irequest.go
index b231db46..1b819874 100644
--- a/ziface/irequest.go
+++ b/ziface/irequest.go
@@ -1,11 +1,12 @@
// Package ziface 主要提供zinx全部抽象层接口定义.
// 包括:
-// IServer 服务mod接口
-// IRouter 路由mod接口
-// IConnection 连接mod层接口
-// IMessage 消息mod接口
-// IDataPack 消息拆解接口
-// IMsgHandler 消息处理及协程池接口
+//
+// IServer 服务mod接口
+// IRouter 路由mod接口
+// IConnection 连接mod层接口
+// IMessage 消息mod接口
+// IDataPack 消息拆解接口
+// IMsgHandler 消息处理及协程池接口
//
// 当前文件描述:
// @Title irequest.go
@@ -16,14 +17,19 @@ package ziface
type HandleStep int
/*
- IRequest 接口：
- 实际上是把客户端请求的链接信息 和 请求的数据 包装到了 Request里
+IRequest 接口：
+实际上是把客户端请求的链接信息 和 请求的数据 包装到了 Request里
*/
type IRequest interface {
GetConnection() IConnection //获取请求连接信息
GetData() []byte //获取请求消息的数据
GetMsgID() uint32 //获取请求的消息ID
+ GetMessage() IMessage //获取请求消息的原始数据 add by uuxia 2023-03-10
+
+ GetResponse() Response //获取解析完后序列化数据
+ SetResponse(Response) //设置解析完后序列化数据
+
BindRouter(router IRouter) //绑定这次请求由哪个路由处理
Call() //转进到下一个处理器开始执行 但是调用此方法的函数会根据先后顺序逆序执行
Abort() //终止处理函数的运行 但调用此方法的函数会执行完毕
diff --git a/ziface/iserver.go b/ziface/iserver.go
index ac81401e..2c53fff0 100644
--- a/ziface/iserver.go
+++ b/ziface/iserver.go
@@ -1,11 +1,12 @@
// Package ziface 主要提供zinx全部抽象层接口定义.
// 包括:
-// IServer 服务mod接口
-// IRouter 路由mod接口
-// IConnection 连接mod层接口
-// IMessage 消息mod接口
-// IDataPack 消息拆解接口
-// IMsgHandler 消息处理及协程池接口
+//
+// IServer 服务mod接口
+// IRouter 路由mod接口
+// IConnection 连接mod层接口
+// IMessage 消息mod接口
+// IDataPack 消息拆解接口
+// IMsgHandler 消息处理及协程池接口
//
// 当前文件描述:
// @Title iserver.go
@@ -13,9 +14,11 @@
// @Author Aceld - Thu Mar 11 10:32:29 CST 2019
package ziface
-import "time"
+import (
+ "time"
+)
-//定义服务接口
+// 定义服务接口
type IServer interface {
Start() //启动服务器方法
Stop() //停止服务器方法
@@ -31,4 +34,6 @@ type IServer interface {
SetPacket(IDataPack) //设置Server绑定的数据协议封包方式
StartHeartBeat(time.Duration) //启动心跳检测
StartHeartBeatWithOption(time.Duration, *HeartBeatOption) //启动心跳检测(自定义回调)
+ AddInterceptor(interceptor Interceptor) //添加协议解析拦截器
+ GetLengthField() LengthField
}
diff --git a/znet/client.go b/znet/client.go
index b1fd0799..1dbbd645 100644
--- a/znet/client.go
+++ b/znet/client.go
@@ -5,6 +5,7 @@ import (
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/zpack"
+ "math"
"net"
"time"
)
@@ -26,6 +27,8 @@ type Client struct {
exitChan chan struct{}
//消息管理模块
msgHandler ziface.IMsgHandle
+ //断粘包解码器
+ LengthField ziface.LengthField
}
func NewClient(ip string, port int, opts ...ClientOption) ziface.IClient {
@@ -35,6 +38,28 @@ func NewClient(ip string, port int, opts ...ClientOption) ziface.IClient {
Port: port,
msgHandler: NewMsgHandle(),
packet: zpack.Factory().NewPack(ziface.ZinxDataPack), //默认使用zinx的TLV封包方式
+ // +---------------+---------------+---------------+
+ // | Tag | Length | Value |
+ // | uint32(4byte) | uint32(4byte) | n byte |
+ // +---------------+---------------+---------------+
+ // Tag： uint32类型，占4字节
+ // Length：uint32类型，占4字节，Length标记Value长度
+ // Value： 占n字节
+ //
+ //说明:
+ // lengthFieldOffset = 4 (Length的字节位索引下标是4) 长度字段的偏差
+ // lengthFieldLength = 4 (Length是4个byte) 长度字段占的字节数
+ // lengthAdjustment = 0 (Length只表示Value长度，程序只会读取Length个字节就结束，后面没有来，故为0，若Value后面还有crc占2字节的话，那么此处就是2。若Length标记的是Tag+Length+Value总长度，那么此处是-8)
+ // initialBytesToStrip = 0 (这个0表示返回完整的协议内容Tag+Length+Value，如果只想返回Value内容，去掉Tag的4字节和Length的4字节，此处就是8) 从解码帧中第一次去除的字节数
+ // maxFrameLength = 2^32 + 4 + 4 (Length为uint32类型，故2^32次方表示Value最大长度，此外Tag和Length各占4字节)
+ //默认使用TLV封包方式
+ LengthField: ziface.LengthField{
+ MaxFrameLength: math.MaxUint32 + 4 + 4,
+ LengthFieldOffset: 4,
+ LengthFieldLength: 4,
+ LengthAdjustment: 0,
+ InitialBytesToStrip: 0,
+ },
}
//应用Option设置
@@ -45,7 +70,15 @@ func NewClient(ip string, port int, opts ...ClientOption) ziface.IClient {
return c
}
-//启动客户端，发送请求且建立链接
+func (this *Client) AddInterceptor(interceptor ziface.Interceptor) {
+ this.msgHandler.AddInterceptor(interceptor)
+}
+
+func (this *Client) GetLengthField() ziface.LengthField {
+ return this.LengthField
+}
+
+// 启动客户端，发送请求且建立链接
func (c *Client) Start() {
c.exitChan = make(chan struct{})
@@ -82,7 +115,7 @@ func (c *Client) Start() {
}()
}
-//启动心跳检测
+// 启动心跳检测
func (c *Client) StartHeartBeat(interval time.Duration) {
checker := NewHeartbeatCheckerC(interval, c)
@@ -92,7 +125,7 @@ func (c *Client) StartHeartBeat(interval time.Duration) {
go checker.Start()
}
-//启动心跳检测(自定义回调)
+// 启动心跳检测(自定义回调)
func (c *Client) StartHeartBeatWithOption(interval time.Duration, option *ziface.HeartBeatOption) {
checker := NewHeartbeatCheckerC(interval, c)
@@ -121,32 +154,32 @@ func (c *Client) Conn() ziface.IConnection {
return c.conn
}
-//设置该Client的连接创建时Hook函数
+// 设置该Client的连接创建时Hook函数
func (c *Client) SetOnConnStart(hookFunc func(ziface.IConnection)) {
c.onConnStart = hookFunc
}
-//设置该Client的连接断开时的Hook函数
+// 设置该Client的连接断开时的Hook函数
func (c *Client) SetOnConnStop(hookFunc func(ziface.IConnection)) {
c.onConnStop = hookFunc
}
-//GetOnConnStart 得到该Server的连接创建时Hook函数
+// GetOnConnStart 得到该Server的连接创建时Hook函数
func (c *Client) GetOnConnStart() func(ziface.IConnection) {
return c.onConnStart
}
-//得到该Server的连接断开时的Hook函数
+// 得到该Server的连接断开时的Hook函数
func (c *Client) GetOnConnStop() func(ziface.IConnection) {
return c.onConnStop
}
-//获取Client绑定的数据协议封包方式
+// 获取Client绑定的数据协议封包方式
func (c *Client) GetPacket() ziface.IDataPack {
return c.packet
}
-//设置Client绑定的数据协议封包方式
+// 设置Client绑定的数据协议封包方式
func (c *Client) SetPacket(packet ziface.IDataPack) {
c.packet = packet
}
diff --git a/znet/connection.go b/znet/connection.go
index 41fb3baa..d7034b53 100644
--- a/znet/connection.go
+++ b/znet/connection.go
@@ -1,14 +1,15 @@
/*
- 服务端Server的链接模块
+服务端Server的链接模块
*/
package znet
import (
"context"
+ "encoding/hex"
"errors"
+ "github.com/aceld/zinx/zcode"
"github.com/aceld/zinx/zlog"
"github.com/aceld/zinx/zpack"
- "io"
"net"
"sync"
"time"
@@ -17,8 +18,8 @@ import (
"github.com/aceld/zinx/ziface"
)
-//Connection Tcp连接模块
-//用于处理Tcp连接的读写业务 一个连接对应一个Connection
+// Connection Tcp连接模块
+// 用于处理Tcp连接的读写业务 一个连接对应一个Connection
type Connection struct {
//当前连接的socket TCP套接字
conn net.Conn
@@ -49,18 +50,21 @@ type Connection struct {
packet ziface.IDataPack
//最后一次活动时间
lastActivityTime time.Time
+ //断粘包解码器
+ lengthFieldDecoder ziface.IDecoder
}
-//newServerConn :for Server, 创建一个Server服务端特性的连接的方法
-//Note: 名字由 NewConnection 更变
+// newServerConn :for Server, 创建一个Server服务端特性的连接的方法
+// Note: 名字由 NewConnection 更变
func newServerConn(server ziface.IServer, conn net.Conn, connID uint32) *Connection {
//初始化Conn属性
c := &Connection{
- conn: conn,
- connID: connID,
- isClosed: false,
- msgBuffChan: nil,
- property: nil,
+ conn: conn,
+ connID: connID,
+ isClosed: false,
+ msgBuffChan: nil,
+ property: nil,
+ lengthFieldDecoder: zcode.NewLengthFieldFrameDecoderByLengthField(server.GetLengthField()),
}
//从server继承过来的属性
@@ -77,14 +81,15 @@ func newServerConn(server ziface.IServer, conn net.Conn, connID uint32) *Connect
return c
}
-//newClientConn :for Client, 创建一个Client服务端特性的连接的方法
+// newClientConn :for Client, 创建一个Client服务端特性的连接的方法
func newClientConn(client ziface.IClient, conn net.Conn) *Connection {
c := &Connection{
- conn: conn,
- connID: 0, //client ignore
- isClosed: false,
- msgBuffChan: nil,
- property: nil,
+ conn: conn,
+ connID: 0, //client ignore
+ isClosed: false,
+ msgBuffChan: nil,
+ property: nil,
+ lengthFieldDecoder: zcode.NewLengthFieldFrameDecoderByLengthField(client.GetLengthField()),
}
//从client继承过来的属性
@@ -96,7 +101,7 @@ func newClientConn(client ziface.IClient, conn net.Conn) *Connection {
return c
}
-//StartWriter 写消息Goroutine， 用户将数据发送给客户端
+// StartWriter 写消息Goroutine， 用户将数据发送给客户端
func (c *Connection) StartWriter() {
zlog.Ins().InfoF("Writer Goroutine is running")
defer zlog.Ins().InfoF("%s [conn Writer exit!]", c.RemoteAddr().String())
@@ -123,7 +128,7 @@ func (c *Connection) StartWriter() {
}
}
-//StartReader 读消息Goroutine，用于从客户端中读取数据
+// StartReader 读消息Goroutine，用于从客户端中读取数据
func (c *Connection) StartReader() {
zlog.Ins().InfoF("[Reader Goroutine is running]")
defer zlog.Ins().InfoF("%s [conn Reader exit!]", c.RemoteAddr().String())
@@ -137,7 +142,7 @@ func (c *Connection) StartReader() {
default:
//读取客户端的Msg head
- headData := make([]byte, c.packet.GetHeadLen())
+ /*headData := make([]byte, c.packet.GetHeadLen())
if _, err := io.ReadFull(c.conn, headData); err != nil {
zlog.Ins().ErrorF("read msg head error %s", err)
return
@@ -175,12 +180,35 @@ func (c *Connection) StartReader() {
} else {
//从绑定好的消息和对应的处理方法中执行对应的Handle方法
go c.msgHandler.DoMsgHandler(req)
+ }*/
+
+ //add by uuxia 2023-02-03
+ buffer := make([]byte, 1024)
+ n, err := c.conn.Read(buffer[:])
+ if err != nil {
+ zlog.Ins().ErrorF("read msg head error%d %s", n, err)
+ return
}
+ zlog.Ins().DebugF("read buffer %s \n", hex.EncodeToString(buffer[0:n]))
+
+ if c.lengthFieldDecoder != nil {
+ bufArrays := c.lengthFieldDecoder.Decode(buffer[0:n])
+ if bufArrays != nil {
+ for _, bytes := range bufArrays {
+ zlog.Ins().DebugF("read buffer %s \n", hex.EncodeToString(bytes))
+ msg := &zpack.Message{DataLen: uint32(len(bytes)), Data: bytes}
+ //得到当前客户端请求的Request数据
+ req := NewRequest(c, msg)
+ c.msgHandler.Decode(req)
+ }
+ }
+ }
+
}
}
}
-//Start 启动连接，让当前连接开始工作
+// Start 启动连接，让当前连接开始工作
func (c *Connection) Start() {
c.ctx, c.cancel = context.WithCancel(context.Background())
//按照用户传递进来的创建连接时需要处理的业务，执行钩子方法
@@ -195,7 +223,7 @@ func (c *Connection) Start() {
}
}
-//Stop 停止连接，结束当前连接状态M
+// Stop 停止连接，结束当前连接状态M
func (c *Connection) Stop() {
c.cancel()
}
@@ -204,22 +232,74 @@ func (c *Connection) GetConnection() net.Conn {
return c.conn
}
-//GetConnID 获取当前连接ID
+// GetConnID 获取当前连接ID
func (c *Connection) GetConnID() uint32 {
return c.connID
}
-//RemoteAddr 获取链接远程地址信息
+// RemoteAddr 获取链接远程地址信息
func (c *Connection) RemoteAddr() net.Addr {
return c.conn.RemoteAddr()
}
-//LocalAddr 获取链接本地地址信息
+// LocalAddr 获取链接本地地址信息
func (c *Connection) LocalAddr() net.Addr {
return c.conn.LocalAddr()
}
-//SendMsg 直接将Message数据发送数据给远程的TCP客户端
+func (c *Connection) Send(data []byte) error {
+ c.msgLock.RLock()
+ defer c.msgLock.RUnlock()
+ if c.isClosed == true {
+ return errors.New("connection closed when send msg")
+ }
+
+ //写回客户端
+ _, err := c.conn.Write(data)
+ if err != nil {
+ zlog.Ins().ErrorF("SendMsg err data = %+v, err = %+v", data, err)
+ return err
+ }
+
+ //写对端成功, 更新链接活动时间
+ c.updateActivity()
+
+ return nil
+}
+
+func (c *Connection) SendToQueue(data []byte) error {
+ c.msgLock.RLock()
+ defer c.msgLock.RUnlock()
+
+ if c.msgBuffChan == nil {
+ c.msgBuffChan = make(chan []byte, utils.GlobalObject.MaxMsgChanLen)
+ //开启用于写回客户端数据流程的Goroutine
+ //此方法只读取MsgBuffChan中的数据没调用SendBuffMsg可以分配内存和启用协程
+ go c.StartWriter()
+ }
+
+ idleTimeout := time.NewTimer(5 * time.Millisecond)
+ defer idleTimeout.Stop()
+
+ if c.isClosed == true {
+ return errors.New("Connection closed when send buff msg")
+ }
+
+ if data == nil {
+ zlog.Ins().ErrorF("Pack data is nil")
+ return errors.New("Pack data is nil")
+ }
+
+ // 发送超时
+ select {
+ case <-idleTimeout.C:
+ return errors.New("send buff msg timeout")
+ case c.msgBuffChan <- data:
+ return nil
+ }
+}
+
+// SendMsg 直接将Message数据发送数据给远程的TCP客户端
func (c *Connection) SendMsg(msgID uint32, data []byte) error {
c.msgLock.RLock()
defer c.msgLock.RUnlock()
@@ -247,7 +327,7 @@ func (c *Connection) SendMsg(msgID uint32, data []byte) error {
return nil
}
-//SendBuffMsg 发生BuffMsg
+// SendBuffMsg 发生BuffMsg
func (c *Connection) SendBuffMsg(msgID uint32, data []byte) error {
c.msgLock.RLock()
defer c.msgLock.RUnlock()
@@ -282,7 +362,7 @@ func (c *Connection) SendBuffMsg(msgID uint32, data []byte) error {
}
}
-//SetProperty 设置链接属性
+// SetProperty 设置链接属性
func (c *Connection) SetProperty(key string, value interface{}) {
c.propertyLock.Lock()
defer c.propertyLock.Unlock()
@@ -293,7 +373,7 @@ func (c *Connection) SetProperty(key string, value interface{}) {
c.property[key] = value
}
-//GetProperty 获取链接属性
+// GetProperty 获取链接属性
func (c *Connection) GetProperty(key string) (interface{}, error) {
c.propertyLock.Lock()
defer c.propertyLock.Unlock()
@@ -305,7 +385,7 @@ func (c *Connection) GetProperty(key string) (interface{}, error) {
return nil, errors.New("no property found")
}
-//RemoveProperty 移除链接属性
+// RemoveProperty 移除链接属性
func (c *Connection) RemoveProperty(key string) {
c.propertyLock.Lock()
defer c.propertyLock.Unlock()
@@ -313,7 +393,7 @@ func (c *Connection) RemoveProperty(key string) {
delete(c.property, key)
}
-//返回ctx，用于用户自定义的go程获取连接退出状态
+// 返回ctx，用于用户自定义的go程获取连接退出状态
func (c *Connection) Context() context.Context {
return c.ctx
}
@@ -348,7 +428,7 @@ func (c *Connection) finalizer() {
zlog.Ins().InfoF("Conn Stop()...ConnID = %d", c.connID)
}
-//callOnConnStart 调用连接OnConnStart Hook函数
+// callOnConnStart 调用连接OnConnStart Hook函数
func (c *Connection) callOnConnStart() {
if c.onConnStart != nil {
zlog.Ins().InfoF("ZINX CallOnConnStart....")
@@ -356,7 +436,7 @@ func (c *Connection) callOnConnStart() {
}
}
-//callOnConnStart 调用连接OnConnStop Hook函数
+// callOnConnStart 调用连接OnConnStop Hook函数
func (c *Connection) callOnConnStop() {
if c.onConnStop != nil {
zlog.Ins().InfoF("ZINX CallOnConnStop....")
diff --git a/znet/msghandler.go b/znet/msghandler.go
index 8b9fa9a6..3b7b9dc3 100644
--- a/znet/msghandler.go
+++ b/znet/msghandler.go
@@ -1,8 +1,10 @@
package znet
import (
+ "encoding/hex"
"fmt"
"github.com/aceld/zinx/utils"
+ "github.com/aceld/zinx/zcode"
"github.com/aceld/zinx/ziface"
"github.com/aceld/zinx/zlog"
)
@@ -12,19 +14,45 @@ type MsgHandle struct {
Apis map[uint32]ziface.IRouter //存放每个MsgID 所对应的处理方法的map属性
WorkerPoolSize uint32 //业务工作Worker池的数量
TaskQueue []chan ziface.IRequest //Worker负责取任务的消息队列
+ builder ziface.InterceptorBuilder //责任链构造器
}
-//NewMsgHandle 创建MsgHandle
+// NewMsgHandle 创建MsgHandle
func NewMsgHandle() *MsgHandle {
return &MsgHandle{
Apis: make(map[uint32]ziface.IRouter),
WorkerPoolSize: utils.GlobalObject.WorkerPoolSize,
//一个worker对应一个queue
TaskQueue: make([]chan ziface.IRequest, utils.GlobalObject.WorkerPoolSize),
+ builder: zcode.NewInterceptorBuilder(),
}
}
-//SendMsgToTaskQueue 将消息交给TaskQueue,由worker进行处理
+func (this *MsgHandle) Intercept(chain ziface.Chain) ziface.Response {
+ request := chain.Request()
+ if request != nil {
+ switch request.(type) {
+ case ziface.IRequest:
+ iRequest := request.(ziface.IRequest)
+ if utils.GlobalObject.WorkerPoolSize > 0 {
+ //已经启动工作池机制，将消息交给Worker处理
+ this.SendMsgToTaskQueue(iRequest)
+ } else {
+ //从绑定好的消息和对应的处理方法中执行对应的Handle方法
+ go this.DoMsgHandler(iRequest)
+ }
+ }
+ }
+ return chain.Proceed(chain.Request())
+}
+
+func (this *MsgHandle) AddInterceptor(interceptor ziface.Interceptor) {
+ if this.builder != nil {
+ this.builder.AddInterceptor(interceptor)
+ }
+}
+
+// SendMsgToTaskQueue 将消息交给TaskQueue,由worker进行处理
func (mh *MsgHandle) SendMsgToTaskQueue(request ziface.IRequest) {
//根据ConnID来分配当前的连接应该由哪个worker负责处理
//轮询的平均分配法则
@@ -34,22 +62,28 @@ func (mh *MsgHandle) SendMsgToTaskQueue(request ziface.IRequest) {
//zlog.Ins().DebugF("Add ConnID=%d request msgID=%d to workerID=%d", request.GetConnection().GetConnID(), request.GetMsgID(), workerID)
//将请求消息发送给任务队列
mh.TaskQueue[workerID] <- request
+ zlog.Ins().ErrorF("SendMsgToTaskQueue-->%s", hex.EncodeToString(request.GetData()))
}
-//DoMsgHandler 马上以非阻塞方式处理消息
+// DoMsgHandler 马上以非阻塞方式处理消息
func (mh *MsgHandle) DoMsgHandler(request ziface.IRequest) {
handler, ok := mh.Apis[request.GetMsgID()]
if !ok {
zlog.Ins().ErrorF("api msgID = %d is not FOUND!", request.GetMsgID())
return
}
+
//Request请求绑定Router对应关系
request.BindRouter(handler)
//执行对应处理方法
request.Call()
}
-//AddRouter 为消息添加具体的处理逻辑
+func (mh *MsgHandle) Decode(request ziface.IRequest) {
+ mh.builder.Execute(request) //将消息丢到责任链，通过责任链里拦截器层层处理层层传递
+}
+
+// AddRouter 为消息添加具体的处理逻辑
func (mh *MsgHandle) AddRouter(msgID uint32, router ziface.IRouter) {
//1 判断当前msg绑定的API处理方法是否已经存在
if _, ok := mh.Apis[msgID]; ok {
@@ -61,7 +95,7 @@ func (mh *MsgHandle) AddRouter(msgID uint32, router ziface.IRouter) {
zlog.Ins().InfoF("Add api msgID = %d", msgID)
}
-//StartOneWorker 启动一个Worker工作流程
+// StartOneWorker 启动一个Worker工作流程
func (mh *MsgHandle) StartOneWorker(workerID int, taskQueue chan ziface.IRequest) {
zlog.Ins().InfoF("Worker ID = %d is started.", workerID)
//不断的等待队列中的消息
@@ -74,8 +108,10 @@ func (mh *MsgHandle) StartOneWorker(workerID int, taskQueue chan ziface.IRequest
}
}
-//StartWorkerPool 启动worker工作池
+// StartWorkerPool 启动worker工作池
func (mh *MsgHandle) StartWorkerPool() {
+ //此处必须把 msghandler 添加到责任链中，并且是责任链最后一环，在msghandler中进行解码后由router做数据分发
+ mh.AddInterceptor(mh)
//遍历需要启动worker的数量，依此启动
for i := 0; i < int(mh.WorkerPoolSize); i++ {
//一个worker被启动
diff --git a/znet/request.go b/znet/request.go
index 79f7a22f..8095b2c2 100644
--- a/znet/request.go
+++ b/znet/request.go
@@ -13,8 +13,9 @@ const (
HANDLE_OVER
)
-//Request 请求
+// Request 请求
type Request struct {
+ response ziface.Response
conn ziface.IConnection //已经和客户端建立好的 链接
msg ziface.IMessage //客户端请求的数据
router ziface.IRouter //请求处理的函数
@@ -23,6 +24,14 @@ type Request struct {
needNext bool
}
+func (r *Request) GetResponse() ziface.Response {
+ return r.response
+}
+
+func (r *Request) SetResponse(response ziface.Response) {
+ r.response = response
+}
+
func NewRequest(conn ziface.IConnection, msg ziface.IMessage) *Request {
req := new(Request)
req.steps = PRE_HANDLE
@@ -34,17 +43,22 @@ func NewRequest(conn ziface.IConnection, msg ziface.IMessage) *Request {
return req
}
-//GetConnection 获取请求连接信息
+// GetMessage 获取消息实体
+func (r *Request) GetMessage() ziface.IMessage {
+ return r.msg
+}
+
+// GetConnection 获取请求连接信息
func (r *Request) GetConnection() ziface.IConnection {
return r.conn
}
-//GetData 获取请求消息的数据
+// GetData 获取请求消息的数据
func (r *Request) GetData() []byte {
return r.msg.GetData()
}
-//GetMsgID 获取请求的消息的ID
+// GetMsgID 获取请求的消息的ID
func (r *Request) GetMsgID() uint32 {
return r.msg.GetMsgID()
}
@@ -89,6 +103,8 @@ func (r *Request) Call() {
r.next()
}
+
+ r.steps = PRE_HANDLE
}
func (r *Request) Abort() {
diff --git a/znet/server.go b/znet/server.go
index 700a12ab..d986591c 100644
--- a/znet/server.go
+++ b/znet/server.go
@@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"github.com/aceld/zinx/zlog"
+ "math"
"net"
"os"
"os/signal"
@@ -27,7 +28,7 @@ var topLine = `┌────────────────────
var borderLine = `│`
var bottomLine = `└──────────────────────────────────────────────────────┘`
-//Server 接口实现，定义一个Server服务类
+// Server 接口实现，定义一个Server服务类
type Server struct {
//服务器的名称
Name string
@@ -49,9 +50,15 @@ type Server struct {
packet ziface.IDataPack
//异步捕获链接关闭状态
exitChan chan struct{}
+ //断粘包解码器
+ LengthField ziface.LengthField
}
-//NewServer 创建一个服务器句柄
+func (this *Server) GetLengthField() ziface.LengthField {
+ return this.LengthField
+}
+
+// NewServer 创建一个服务器句柄
func NewServer(opts ...Option) ziface.IServer {
printLogo()
@@ -65,6 +72,28 @@ func NewServer(opts ...Option) ziface.IServer {
exitChan: nil,
//默认使用zinx的TLV封包方式
packet: zpack.Factory().NewPack(ziface.ZinxDataPack),
+ // +---------------+---------------+---------------+
+ // | Tag | Length | Value |
+ // | uint32(4byte) | uint32(4byte) | n byte |
+ // +---------------+---------------+---------------+
+ // Tag： uint32类型，占4字节
+ // Length：uint32类型，占4字节，Length标记Value长度
+ // Value： 占n字节
+ //
+ //说明:
+ // lengthFieldOffset = 4 (Length的字节位索引下标是4) 长度字段的偏差
+ // lengthFieldLength = 4 (Length是4个byte) 长度字段占的字节数
+ // lengthAdjustment = 0 (Length只表示Value长度，程序只会读取Length个字节就结束，后面没有来，故为0，若Value后面还有crc占2字节的话，那么此处就是2。若Length标记的是Tag+Length+Value总长度，那么此处是-8)
+ // initialBytesToStrip = 0 (这个0表示返回完整的协议内容Tag+Length+Value，如果只想返回Value内容，去掉Tag的4字节和Length的4字节，此处就是8) 从解码帧中第一次去除的字节数
+ // maxFrameLength = 2^32 + 4 + 4 (Length为uint32类型，故2^32次方表示Value最大长度，此外Tag和Length各占4字节)
+ //默认使用TLV封包方式
+ LengthField: ziface.LengthField{
+ MaxFrameLength: math.MaxUint32 + 4 + 4,
+ LengthFieldOffset: 4,
+ LengthFieldLength: 4,
+ LengthAdjustment: 0,
+ InitialBytesToStrip: 0,
+ },
}
for _, opt := range opts {
@@ -77,7 +106,7 @@ func NewServer(opts ...Option) ziface.IServer {
return s
}
-//NewServer 创建一个服务器句柄
+// NewServer 创建一个服务器句柄
func NewUserConfServer(config *utils.Config, opts ...Option) ziface.IServer {
//打印logo
printLogo()
@@ -91,6 +120,14 @@ func NewUserConfServer(config *utils.Config, opts ...Option) ziface.IServer {
ConnMgr: NewConnManager(),
exitChan: nil,
packet: zpack.Factory().NewPack(ziface.ZinxDataPack),
+ //默认使用TLV封包方式
+ LengthField: ziface.LengthField{
+ MaxFrameLength: math.MaxUint32 + 4 + 4,
+ LengthFieldOffset: 4,
+ LengthFieldLength: 4,
+ LengthAdjustment: 0,
+ InitialBytesToStrip: 0,
+ },
}
//更替打包方式
for _, opt := range opts {
@@ -107,7 +144,11 @@ func NewUserConfServer(config *utils.Config, opts ...Option) ziface.IServer {
//============== 实现 ziface.IServer 里的全部接口方法 ========
-//Start 开启网络服务
+func (this *Server) AddInterceptor(interceptor ziface.Interceptor) {
+ this.msgHandler.AddInterceptor(interceptor)
+}
+
+// Start 开启网络服务
func (s *Server) Start() {
zlog.Ins().InfoF("[START] Server name: %s,listenner at IP: %s, Port %d is starting", s.Name, s.IP, s.Port)
s.exitChan = make(chan struct{})
@@ -181,7 +222,7 @@ func (s *Server) Start() {
}()
}
-//Stop 停止服务
+// Stop 停止服务
func (s *Server) Stop() {
zlog.Ins().InfoF("[STOP] Zinx server , name %s", s.Name)
@@ -191,7 +232,7 @@ func (s *Server) Stop() {
close(s.exitChan)
}
-//Serve 运行服务
+// Serve 运行服务
func (s *Server) Serve() {
s.Start()
@@ -206,32 +247,32 @@ func (s *Server) Serve() {
zlog.Ins().InfoF("[SERVE] Zinx server , name %s, Serve Interrupt, signal = %v", s.Name, sig)
}
-//AddRouter 路由功能：给当前服务注册一个路由业务方法，供客户端链接处理使用
+// AddRouter 路由功能：给当前服务注册一个路由业务方法，供客户端链接处理使用
func (s *Server) AddRouter(msgID uint32, router ziface.IRouter) {
s.msgHandler.AddRouter(msgID, router)
}
-//GetConnMgr 得到链接管理
+// GetConnMgr 得到链接管理
func (s *Server) GetConnMgr() ziface.IConnManager {
return s.ConnMgr
}
-//SetOnConnStart 设置该Server的连接创建时Hook函数
+// SetOnConnStart 设置该Server的连接创建时Hook函数
func (s *Server) SetOnConnStart(hookFunc func(ziface.IConnection)) {
s.onConnStart = hookFunc
}
-//SetOnConnStop 设置该Server的连接断开时的Hook函数
+// SetOnConnStop 设置该Server的连接断开时的Hook函数
func (s *Server) SetOnConnStop(hookFunc func(ziface.IConnection)) {
s.onConnStop = hookFunc
}
-//GetOnConnStart 得到该Server的连接创建时Hook函数
+// GetOnConnStart 得到该Server的连接创建时Hook函数
func (s *Server) GetOnConnStart() func(ziface.IConnection) {
return s.onConnStart
}
-//得到该Server的连接断开时的Hook函数
+// 得到该Server的连接断开时的Hook函数
func (s *Server) GetOnConnStop() func(ziface.IConnection) {
return s.onConnStop
}