Event Processing and Parsers ​

type IWorker interface {
    Write(event.IEventStruct) error      // Add event to processing queue
    GetUUID() string                      // Return worker's UUID
    GetDestroyUUID() uint64               // Return socket ID for lifecycle tracking
    IfUsed() bool                         // Check if worker is currently referenced
    Get()                                 // Increment reference count
    Put()                                 // Decrement reference count
    CloseEventWorker()                    // Signal worker to close
}

type IParser interface {
    detect(b []byte) error        // Attempt to identify protocol from payload
    Write(b []byte) (int, error)  // Accumulate payload data
    ParserType() ParserType       // Return parser type identifier
    PacketType() PacketType       // Return packet encoding (gzip, etc.)
    Name() string                 // Return human-readable parser name
    IsDone() bool                 // Check if parsing is complete
    Init()                        // Initialize parser state
    Display() []byte              // Generate formatted output
    Reset()                       // Reset parser for reuse
}

func init() {
    hr := &HTTPRequest{}
    hr.Init()
    Register(hr)  // Adds to global parsers map
}

func (hr *HTTPRequest) detect(payload []byte) error {
    rd := bytes.NewReader(payload)
    buf := bufio.NewReader(rd)
    _, err := http.ReadRequest(buf)
    if err != nil {
        return err  // Not a valid HTTP request
    }
    return nil  // Valid HTTP request detected
}

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  // Not a valid HTTP response
    }
    return nil  // Valid HTTP response detected
}

// 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  // Embed base event metadata
    CustomField1 uint32
    CustomField2 [64]byte
}

func (e *MyCustomEvent) Decode(payload []byte) error {
    // Deserialize from eBPF C struct layout
    if len(payload) < 104 {  // sizeof(Base) + custom fields
        return errors.New("payload too short")
    }
    
    // Decode base fields
    e.Base.Decode(payload)
    
    // Decode custom fields
    offset := 40  // Size of Base struct
    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  // Route to EventProcessor
}

func (e *MyCustomEvent) GetUUID() string {
    // Generate unique identifier for connection grouping
    return fmt.Sprintf("%d_%d_%s_%d", 
        e.PID, e.TID, e.PName, e.CustomField1)
}

// Implement remaining IEventStruct methods...

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
}

// In module's Run() method
for {
    select {
    case event := <-eventChannel:
        m.processor.Write(event)  // Routes to 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
    // Protocol-specific fields
    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 {
    // Check for protocol magic bytes or structure
    if len(payload) < 4 {
        return errors.New("payload too short")
    }
    
    // Example: Check for magic number
    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) {
    // First write: parse header
    if !p.isInit {
        n, err := p.reader.Write(b)
        if err != nil {
            return n, err
        }
        
        // Attempt to parse header
        err = p.parseHeader()
        if err != nil {
            return 0, err
        }
        
        p.isInit = true
        return n, nil
    }
    
    // Subsequent writes: accumulate body
    n, err := p.reader.Write(b)
    if err != nil {
        return n, err
    }
    
    // Check if complete based on protocol length field
    if p.reader.Len() >= p.header.TotalLength {
        p.isDone = true
    }
    
    return n, nil
}

func (p *MyProtocolParser) Display() []byte {
    var output bytes.Buffer
    
    // Format header
    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")
    
    // Format body
    body := p.reader.Bytes()[p.header.HeaderSize:]
    output.Write(body)
    
    return output.Bytes()
}

func (p *MyProtocolParser) ParserType() ParserType {
    return ParserTypeNull  // Or define new ParserType constant
}

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)  // Makes parser available for detection
}

func (p *MyProtocolParser) Display() []byte {
    rawData := p.reader.Bytes()
    
    if p.isCompressed {
        reader, err := gzip.NewReader(bytes.NewReader(rawData))
        if err != nil {
            return rawData  // Return raw on error
        }
        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
    }
    
    // Check for chunk terminator
    if bytes.HasSuffix(p.reader.Bytes(), []byte("0\r\n\r\n")) {
        p.isDone = true
    }
    
    return n, nil
}

type HTTPRequest struct {
    request    *http.Request     // Parsed HTTP request
    packerType PacketType        // Compression type
    isDone     bool              // Parsing complete flag
    isInit     bool              // Header parsed flag
    reader     *bytes.Buffer     // Raw data accumulator
    bufReader  *bufio.Reader     // Buffered reader for parsing
}

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()
    
    // Valid payload
    validPayload := []byte{0x12, 0x34, 0x56, 0x78, ...}
    err := p.detect(validPayload)
    if err != nil {
        t.Errorf("detect() failed on valid payload: %v", err)
    }
    
    // Invalid payload
    invalidPayload := []byte{0xFF, 0xFF, 0xFF, 0xFF}
    err = p.detect(invalidPayload)
    if err == nil {
        t.Error("detect() should have failed on invalid payload")
    }
}

func TestMyProtocolParser_WriteAndDisplay(t *testing.T) {
    p := &MyProtocolParser{}
    p.Init()
    
    // Write payload chunks
    chunk1 := []byte{...}
    chunk2 := []byte{...}
    
    n, err := p.Write(chunk1)
    if err != nil {
        t.Fatalf("Write() failed: %v", err)
    }
    
    n, err = p.Write(chunk2)
    if err != nil {
        t.Fatalf("Write() failed: %v", err)
    }
    
    // Check if done
    if !p.IsDone() {
        t.Error("Parser should be done after full payload")
    }
    
    // Get formatted output
    output := p.Display()
    if len(output) == 0 {
        t.Error("Display() returned empty output")
    }
}

func TestMyProtocolParser_Integration(t *testing.T) {
    // Create EventProcessor with test logger
    var buf bytes.Buffer
    logger := &buf
    ep := NewEventProcessor(logger, false, 0)
    
    // Start processor
    go ep.Serve()
    
    // Create test event
    event := &MyTestEvent{
        Base: event.Base{
            PID: 1234,
            // ... other fields
        },
        // ... custom fields
    }
    
    // Write event
    ep.Write(event)
    
    // Wait for processing
    time.Sleep(2 * time.Second)
    
    // Check output
    output := buf.String()
    if !strings.Contains(output, "My Protocol") {
        t.Errorf("Output doesn't contain expected content: %s", output)
    }
    
    // Cleanup
    ep.Close()
}