input

Documentation

Index

• Constants
• Variables
• type FileWatcher
  • func NewFileWatcher(f *file.File) (fw *FileWatcher, err error)
  • func (fw *FileWatcher) Restart() error
  • func (fw *FileWatcher) Start()
• type Input
  • func New(ctx context.Context, conf map[string]*keyword.TaskConfig, states []file.State) (*Input, error)
  • func (client *Input) AddInput(_ <-chan interface{})
  • func (client *Input) AddOutput(_ chan<- interface{})
  • func (client *Input) ID() string
  • func (client *Input) Reload(_ interface{})
  • func (client *Input) Start() error
  • func (client *Input) Stop()
  • func (client *Input) Wait()

Constants

const (
	FilePositionHead = int64(0)
	FilePositionEnd  = int64(-1)
)

Variables

var (
	CounterRead         = uint64(0)  // 计数器，读的行数
	CounterLastOpenFile atomic.Value // 最后一次打开的文件
	ScanTickerDuration  = 5 * time.Second
)

Types

type FileWatcher

type FileWatcher struct {
	FileTail *tail.Tail

	File *file.File
	// contains filtered or unexported fields
}

func NewFileWatcher

func NewFileWatcher(f *file.File) (fw *FileWatcher, err error)

func (*FileWatcher) Restart

func (fw *FileWatcher) Restart() error

func (*FileWatcher) Start

func (fw *FileWatcher) Start()

type Input

type Input struct {
	WatchFiles sync.Map // 所有打开的文件 <filename, *FileWatcher>
	// contains filtered or unexported fields
}

Input task data

var SingleInstance *Input // Input模块

func New

func New(ctx context.Context, conf map[string]*keyword.TaskConfig, states []file.State) (*Input, error)

New construct a new input

func (*Input) AddInput

func (client *Input) AddInput(_ <-chan interface{})

func (*Input) AddOutput

func (client *Input) AddOutput(_ chan<- interface{})

func (*Input) ID

func (client *Input) ID() string

func (*Input) Reload

func (client *Input) Reload(_ interface{})

Reload reload task

func (*Input) Start

func (client *Input) Start() error

func (*Input) Stop

func (client *Input) Stop()

Stop stops the input and with it all harvesters

func (*Input) Wait

func (client *Input) Wait()

Wait wg.wait for task