processor

Documentation

Index

• Variables
• func New(ctx context.Context, cfg keyword.ProcessConfig, taskType string) (module.Module, error)
• type EventProcessor
  • func NewEventProcessor(cfg keyword.ProcessConfig) (*EventProcessor, error)
  • func (client *EventProcessor) Filter(event *module.LogEvent) bool
  • func (client *EventProcessor) Handle(event *module.LogEvent) (interface{}, error)
  • func (client *EventProcessor) Send(event interface{}, outputs []chan<- interface{})
• type IProcessor
• type Processor
  • func (client *Processor) AddInput(input <-chan interface{})
  • func (client *Processor) AddOutput(output chan<- interface{})
  • func (client *Processor) ID() string
  • func (client *Processor) Reload(cfg interface{})
  • func (client *Processor) Start() error
  • func (client *Processor) Stop()
  • func (client *Processor) Wait()

Variables

var CounterProcess = uint64(0) // 计数器，读的行数

Functions

func New

func New(ctx context.Context, cfg keyword.ProcessConfig, taskType string) (module.Module, error)

Types

type EventProcessor

type EventProcessor struct {
	// contains filtered or unexported fields
}

func NewEventProcessor

func NewEventProcessor(cfg keyword.ProcessConfig) (*EventProcessor, error)

func (*EventProcessor) Filter

func (client *EventProcessor) Filter(event *module.LogEvent) bool

func (*EventProcessor) Handle

func (client *EventProcessor) Handle(event *module.LogEvent) (interface{}, error)

func (*EventProcessor) Send

func (client *EventProcessor) Send(event interface{}, outputs []chan<- interface{})

type IProcessor

type IProcessor interface {
	// first return value is nil: drop line
	Handle(event *module.LogEvent) (interface{}, error)
	Send(event interface{}, outputs []chan<- interface{})
}

type Processor

type Processor struct {
	// contains filtered or unexported fields
}

func (*Processor) AddInput

func (client *Processor) AddInput(input <-chan interface{})

AddInput. implement module interface

func (*Processor) AddOutput

func (client *Processor) AddOutput(output chan<- interface{})

AddOutput add one output

func (*Processor) ID

func (client *Processor) ID() string

func (*Processor) Reload

func (client *Processor) Reload(cfg interface{})

Reload

func (*Processor) Start

func (client *Processor) Start() error

func (*Processor) Stop

func (client *Processor) Stop()

Stop stops the input and with it all harvesters

func (*Processor) Wait

func (client *Processor) Wait()

Wait