plugins

Documentation

Index

• Variables
• func RegisterDecoder(name string, decoder func() interface{})
• func RegisterEncoder(name string, Encoder func() interface{})
• func RegisterInput(name string, input func() interface{})
• func RegisterOutput(name string, out func() interface{})
• type Decoder
• type Encoder
• type Input
• type InputRunner
  • func NewInputRunner(in, router chan *PipelinePack) InputRunner
• type MasterConfig
  • func DefaultMasterConfig() (master *MasterConfig)
  • func (self *MasterConfig) IsShuttingDown() (stopping bool)
  • func (self *MasterConfig) ShutDown()
  • func (self *MasterConfig) SigChan() chan os.Signal
• type Message
• type Output
• type OutputRunner
  • func NewOutputRunner(in chan *PipelinePack) OutputRunner
• type Pipeline
  • func NewPipeLine() *Pipeline
  • func (this *Pipeline) LoadConfig(plugConfig map[string]toml.Primitive) error
  • func (this *Pipeline) Run(mc *MasterConfig)
  • func (this *Pipeline) SignalWorker()
• type PipelinePack
  • func NewPipelinePack(recycleChan chan *PipelinePack) (pack *PipelinePack)
  • func PipeDecoder(name string, pack *PipelinePack) (rpack *PipelinePack, err error)
  • func PipeEncoder(name string, pack *PipelinePack) (rpack *PipelinePack, err error)
  • func (this *PipelinePack) Recycle()
  • func (this *PipelinePack) Zero()
• type PluginCommonConfig
• type PluginConfig
• type Router
  • func (self *Router) AddInChan(inChan chan *PipelinePack)
  • func (self *Router) AddOutChan(matchtag string, outChan chan *PipelinePack) error
  • func (self *Router) Init()
  • func (self *Router) Loop()

Variables

var PluginTypeRegex = regexp.MustCompile("(Input|Output|Encoder|Decoder)$")

Functions

func RegisterDecoder

func RegisterDecoder(name string, decoder func() interface{})

func RegisterEncoder

func RegisterEncoder(name string, Encoder func() interface{})

func RegisterInput

func RegisterInput(name string, input func() interface{})

func RegisterOutput

func RegisterOutput(name string, out func() interface{})

Types

type Decoder

type Decoder interface {
	Init(config toml.Primitive) error
	Decode(pack *PipelinePack) (*PipelinePack, error)
}

type Encoder

type Encoder interface {
	Init(config toml.Primitive) error
	Encode(pack *PipelinePack) (*PipelinePack, error)
}

type Input

type Input interface {
	Init(pcf *PluginCommonConfig, config toml.Primitive) error
	Run(in InputRunner) error
}

type InputRunner

type InputRunner interface {
	InChan() chan *PipelinePack
	RouterChan() chan *PipelinePack
	Start(cf toml.Primitive)
}

func NewInputRunner

func NewInputRunner(in, router chan *PipelinePack) InputRunner

type MasterConfig

type MasterConfig struct {
	PoolSize       int
	PluginChanSize int
	MaxMsgLoops    uint

	BaseDir string

	Hostname string
	// contains filtered or unexported fields
}

func DefaultMasterConfig

func DefaultMasterConfig() (master *MasterConfig)

func (*MasterConfig) IsShuttingDown

func (self *MasterConfig) IsShuttingDown() (stopping bool)

func (*MasterConfig) ShutDown

func (self *MasterConfig) ShutDown()

func (*MasterConfig) SigChan

func (self *MasterConfig) SigChan() chan os.Signal

type Message

type Message struct {
	MsgBytes  []byte
	Tag       string
	Timestamp int64
	Data      map[string]interface{}
	sync.RWMutex
}

type Output

type Output interface {
	Init(pcf *PluginCommonConfig, config toml.Primitive) error
	Run(out OutputRunner) error
}

type OutputRunner

type OutputRunner interface {
	InChan() chan *PipelinePack
	Start(cf toml.Primitive)
}

func NewOutputRunner

func NewOutputRunner(in chan *PipelinePack) OutputRunner

type Pipeline

type Pipeline struct {
	InputRunners  []interface{}
	OutputRunners []interface{}
	DecodeRunners []interface{}
	EncodeRunners []interface{}
	// contains filtered or unexported fields
}

func NewPipeLine

func NewPipeLine() *Pipeline

func (*Pipeline) LoadConfig

func (this *Pipeline) LoadConfig(plugConfig map[string]toml.Primitive) error

func (*Pipeline) Run

func (this *Pipeline) Run(mc *MasterConfig)

func (*Pipeline) SignalWorker

func (this *Pipeline) SignalWorker()

type PipelinePack

type PipelinePack struct {
	MsgBytes    []byte
	Msg         Message
	RecycleChan chan *PipelinePack
	RefCount    int32
}

func NewPipelinePack

func NewPipelinePack(recycleChan chan *PipelinePack) (pack *PipelinePack)

func PipeDecoder

func PipeDecoder(name string, pack *PipelinePack) (rpack *PipelinePack, err error)

func PipeEncoder

func PipeEncoder(name string, pack *PipelinePack) (rpack *PipelinePack, err error)

func (*PipelinePack) Recycle

func (this *PipelinePack) Recycle()

func (*PipelinePack) Zero

func (this *PipelinePack) Zero()

type PluginCommonConfig

type PluginCommonConfig struct {
	Type    string `toml:"type"`
	Tag     string `toml:"tag"`
	Decoder string `toml:"decoder"`
	Encoder string `toml:"encoder"`
}

type PluginConfig

type PluginConfig map[string]toml.Primitive

type Router

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

func (*Router) AddInChan

func (self *Router) AddInChan(inChan chan *PipelinePack)

func (*Router) AddOutChan

func (self *Router) AddOutChan(matchtag string, outChan chan *PipelinePack) error

func (*Router) Init

func (self *Router) Init()

func (*Router) Loop

func (self *Router) Loop()