事件处理与解析器 ​

type IWorker interface {
    Write(event.IEventStruct) error      // 向处理队列添加事件
    GetUUID() string                      // 返回 worker 的 UUID
    GetDestroyUUID() uint64               // 返回用于生命周期跟踪的套接字 ID
    IfUsed() bool                         // 检查 worker 当前是否被引用
    Get()                                 // 增加引用计数
    Put()                                 // 减少引用计数
    CloseEventWorker()                    // 发信号通知 worker 关闭
}

type IParser interface {
    detect(b []byte) error        // 尝试从有效载荷识别协议
    Write(b []byte) (int, error)  // 累积有效载荷数据
    ParserType() ParserType       // 返回解析器类型标识符
    PacketType() PacketType       // 返回数据包编码（gzip 等）
    Name() string                 // 返回人类可读的解析器名称
    IsDone() bool                 // 检查解析是否完成
    Init()                        // 初始化解析器状态
    Display() []byte              // 生成格式化输出
    Reset()                       // 重置解析器以供重用
}

func init() {
    hr := &HTTPRequest{}
    hr.Init()
    Register(hr)  // 添加到全局 parsers 映射
}

func (hr *HTTPRequest) detect(payload []byte) error {
    rd := bytes.NewReader(payload)
    buf := bufio.NewReader(rd)
    _, err := http.ReadRequest(buf)
    if err != nil {
        return err  // 不是有效的 HTTP 请求
    }
    return nil  // 检测到有效的 HTTP 请求
}

func (hr *HTTPResponse) detect(payload []byte) error {
    rd := bytes.NewReader(payload)
    buf := bufio.NewReader(rd)
    _, err := http.ReadResponse(buf, nil)
    if err != nil {
        return err  // 不是有效的 HTTP 响应
    }
    return nil  // 检测到有效的 HTTP 响应
}

// user/event/mymodule/my_event.go
package mymodule

import (
    "encoding/binary"
    "github.com/gojue/ecapture/user/event"
    pb "github.com/gojue/ecapture/protobuf/gen/v1"
)

type MyCustomEvent struct {
    event.Base  // 嵌入基础事件元数据
    CustomField1 uint32
    CustomField2 [64]byte
}

func (e *MyCustomEvent) Decode(payload []byte) error {
    // 从 eBPF C 结构布局反序列化
    if len(payload) < 104 {  // sizeof(Base) + custom fields
        return errors.New("payload too short")
    }
    
    // 解码基础字段
    e.Base.Decode(payload)
    
    // 解码自定义字段
    offset := 40  // Base 结构的大小
    e.CustomField1 = binary.LittleEndian.Uint32(payload[offset:])
    copy(e.CustomField2[:], payload[offset+4:offset+68])
    
    return nil
}

func (e *MyCustomEvent) Clone() event.IEventStruct {
    newEvent := *e
    return &newEvent
}

func (e *MyCustomEvent) EventType() event.Type {
    return event.TypeEventProcessor  // 路由到 EventProcessor
}

func (e *MyCustomEvent) GetUUID() string {
    // 生成用于连接分组的唯一标识符
    return fmt.Sprintf("%d_%d_%s_%d", 
        e.PID, e.TID, e.PName, e.CustomField1)
}

// 实现其余的 IEventStruct 方法...

func (m *MyModule) Decode(em *ebpf.Map, b []byte) (event.IEventStruct, error) {
    e := &mymodule.MyCustomEvent{}
    err := e.Decode(b)
    if err != nil {
        return nil, err
    }
    return e, nil
}

// 在模块的 Run() 方法中
for {
    select {
    case event := <-eventChannel:
        m.processor.Write(event)  // 路由到 EventProcessor
    }
}

// pkg/event_processor/my_protocol.go
package event_processor

import (
    "bytes"
    "bufio"
)

type MyProtocolParser struct {
    reader    *bytes.Buffer
    bufReader *bufio.Reader
    isDone    bool
    isInit    bool
    // 协议特定字段
    header    MyProtocolHeader
}

func (p *MyProtocolParser) Init() {
    p.reader = bytes.NewBuffer(nil)
    p.bufReader = bufio.NewReader(p.reader)
    p.isDone = false
    p.isInit = false
}

func (p *MyProtocolParser) Name() string {
    return "MyProtocolParser"
}

func (p *MyProtocolParser) detect(payload []byte) error {
    // 检查协议魔术字节或结构
    if len(payload) < 4 {
        return errors.New("payload too short")
    }
    
    // 示例：检查魔术数字
    magic := binary.BigEndian.Uint32(payload[0:4])
    if magic != MY_PROTOCOL_MAGIC {
        return errors.New("not my protocol")
    }
    
    return nil
}

func (p *MyProtocolParser) Write(b []byte) (int, error) {
    // 第一次写入：解析头部
    if !p.isInit {
        n, err := p.reader.Write(b)
        if err != nil {
            return n, err
        }
        
        // 尝试解析头部
        err = p.parseHeader()
        if err != nil {
            return 0, err
        }
        
        p.isInit = true
        return n, nil
    }
    
    // 后续写入：累积主体
    n, err := p.reader.Write(b)
    if err != nil {
        return n, err
    }
    
    // 根据协议长度字段检查是否完整
    if p.reader.Len() >= p.header.TotalLength {
        p.isDone = true
    }
    
    return n, nil
}

func (p *MyProtocolParser) Display() []byte {
    var output bytes.Buffer
    
    // 格式化头部
    fmt.Fprintf(&output, "My Protocol v%d\n", p.header.Version)
    fmt.Fprintf(&output, "Type: %d\n", p.header.MessageType)
    fmt.Fprintf(&output, "Length: %d\n", p.header.TotalLength)
    fmt.Fprintf(&output, "\n")
    
    // 格式化主体
    body := p.reader.Bytes()[p.header.HeaderSize:]
    output.Write(body)
    
    return output.Bytes()
}

func (p *MyProtocolParser) ParserType() ParserType {
    return ParserTypeNull  // 或定义新的 ParserType 常量
}

func (p *MyProtocolParser) PacketType() PacketType {
    return PacketTypeNull
}

func (p *MyProtocolParser) IsDone() bool {
    return p.isDone
}

func (p *MyProtocolParser) Reset() {
    p.isDone = false
    p.isInit = false
    p.reader.Reset()
    p.bufReader.Reset(p.reader)
}

func init() {
    p := &MyProtocolParser{}
    p.Init()
    Register(p)  // 使解析器可用于检测
}

func (p *MyProtocolParser) Display() []byte {
    rawData := p.reader.Bytes()
    
    if p.isCompressed {
        reader, err := gzip.NewReader(bytes.NewReader(rawData))
        if err != nil {
            return rawData  // 错误时返回原始数据
        }
        defer reader.Close()
        
        decompressed, err := io.ReadAll(reader)
        if err != nil {
            return rawData
        }
        
        p.packerType = PacketTypeGzip
        return decompressed
    }
    
    return rawData
}

func (p *MyProtocolParser) Write(b []byte) (int, error) {
    n, err := p.reader.Write(b)
    if err != nil {
        return n, err
    }
    
    // 检查块终止符
    if bytes.HasSuffix(p.reader.Bytes(), []byte("0\r\n\r\n")) {
        p.isDone = true
    }
    
    return n, nil
}

type HTTPRequest struct {
    request    *http.Request     // 已解析的 HTTP 请求
    packerType PacketType        // 压缩类型
    isDone     bool              // 解析完成标志
    isInit     bool              // 头部已解析标志
    reader     *bytes.Buffer     // 原始数据累加器
    bufReader  *bufio.Reader     // 用于解析的缓冲读取器
}

var parsers = make(map[string]IParser)

func Register(p IParser) {
    if p == nil {
        panic("Register Parser is nil")
    }
    name := p.Name()
    if _, dup := parsers[name]; dup {
        panic(fmt.Sprintf("Register called twice for Parser %s", name))
    }
    parsers[name] = p
}

func init() {
    hr := &HTTPRequest{}
    hr.Init()
    Register(hr)
}

// pkg/event_processor/my_protocol_test.go
package event_processor

import (
    "testing"
)

func TestMyProtocolParser_Detect(t *testing.T) {
    p := &MyProtocolParser{}
    p.Init()
    
    // 有效载荷
    validPayload := []byte{0x12, 0x34, 0x56, 0x78, ...}
    err := p.detect(validPayload)
    if err != nil {
        t.Errorf("detect() 在有效载荷上失败：%v", err)
    }
    
    // 无效载荷
    invalidPayload := []byte{0xFF, 0xFF, 0xFF, 0xFF}
    err = p.detect(invalidPayload)
    if err == nil {
        t.Error("detect() 应该在无效载荷上失败")
    }
}

func TestMyProtocolParser_WriteAndDisplay(t *testing.T) {
    p := &MyProtocolParser{}
    p.Init()
    
    // 写入载荷块
    chunk1 := []byte{...}
    chunk2 := []byte{...}
    
    n, err := p.Write(chunk1)
    if err != nil {
        t.Fatalf("Write() 失败：%v", err)
    }
    
    n, err = p.Write(chunk2)
    if err != nil {
        t.Fatalf("Write() 失败：%v", err)
    }
    
    // 检查是否完成
    if !p.IsDone() {
        t.Error("解析器在完整载荷后应该完成")
    }
    
    // 获取格式化输出
    output := p.Display()
    if len(output) == 0 {
        t.Error("Display() 返回空输出")
    }
}

func TestMyProtocolParser_Integration(t *testing.T) {
    // 使用测试日志记录器创建 EventProcessor
    var buf bytes.Buffer
    logger := &buf
    ep := NewEventProcessor(logger, false, 0)
    
    // 启动处理器
    go ep.Serve()
    
    // 创建测试事件
    event := &MyTestEvent{
        Base: event.Base{
            PID: 1234,
            // ... 其他字段
        },
        // ... 自定义字段
    }
    
    // 写入事件
    ep.Write(event)
    
    // 等待处理
    time.Sleep(2 * time.Second)
    
    // 检查输出
    output := buf.String()
    if !strings.Contains(output, "My Protocol") {
        t.Errorf("输出不包含预期内容：%s", output)
    }
    
    // 清理
    ep.Close()
}