pipeline

Documentation

Index

• Variables
• func AddSourceFields(header map[string]interface{}, fields map[string]interface{}, underRoot bool, ...)
• func GetWithType(category api.Category, typename api.Type, info Info) (api.Component, error)
• func Register(category api.Category, typename api.Type, factory Factory)
• func SetDefaultConfigRaw(defaults Config)
• type Config
  • func GetDefaultConfigRaw() (*Config, error)
  • func (c *Config) DeepCopy() *Config
  • func (c *Config) SetDefaults()
  • func (c *Config) Validate() error
• type Epoch
  • func NewEpoch(pipelineName string) *Epoch
  • func (e *Epoch) Equal(ae *Epoch) bool
  • func (e *Epoch) Increase()
  • func (e *Epoch) IsEmpty() bool
  • func (e *Epoch) String() string
• type Factory
• type Info
• type Pipeline
  • func NewPipeline(pipelineConfig *Config) *Pipeline
  • func (p *Pipeline) Start() error
  • func (p *Pipeline) Stop()
• type RegisterCenter
  • func NewRegisterCenter() *RegisterCenter
  • func (r *RegisterCenter) LoadCodeCategoryComponents(category api.Category) map[string]api.Component
  • func (r *RegisterCenter) LoadCodeComponents() map[string]api.Component
  • func (r *RegisterCenter) LoadCodeInterceptors() map[string]api.Interceptor
  • func (r *RegisterCenter) LoadDefaultQueue() api.Queue
  • func (r *RegisterCenter) LoadInterceptors() []api.Interceptor
  • func (r *RegisterCenter) LoadQueue(typename api.Type, name string) api.Queue
  • func (r *RegisterCenter) LoadQueueListeners() []spi.QueueListener
  • func (r *RegisterCenter) LoadSink(typename api.Type, name string) api.Sink
  • func (r *RegisterCenter) LoadSource(typename api.Type, name string) api.Source
  • func (r *RegisterCenter) LoadWithType(typename api.Type, name string, componentType api.Category) api.Component
  • func (r *RegisterCenter) Register(component api.Component, name string) error
  • func (r *RegisterCenter) RegisterListener(listener spi.ComponentListener)
  • func (r *RegisterCenter) RemoveByCode(code string)

Variables

var (
	ErrPipelineNameRequired   = errors.New("pipelines[n].name is required")
	ErrSourceNameRequired     = errors.New("sources[n].name is required")
	ErrPipelineSourceRequired = errors.New("pipelines[n].source is required")
	ErrPipelineSinkRequired   = errors.New("pipelines[n].sink is required")
)

Functions

func AddSourceFields

func AddSourceFields(header map[string]interface{}, fields map[string]interface{}, underRoot bool, fieldsKey string)

func GetWithType

func GetWithType(category api.Category, typename api.Type, info Info) (api.Component, error)

func Register

func Register(category api.Category, typename api.Type, factory Factory)

func SetDefaultConfigRaw

func SetDefaultConfigRaw(defaults Config)

Types

type Config

type Config struct {
	Name             string        `yaml:"name,omitempty" validate:"required"`
	CleanDataTimeout time.Duration `yaml:"cleanDataTimeout,omitempty" default:"5s"`

	Queue        *queue.Config         `yaml:"queue,omitempty" validate:"dive,required"`
	Interceptors []*interceptor.Config `yaml:"interceptors,omitempty"`
	Sources      []*source.Config      `yaml:"sources,omitempty" validate:"dive,required"`
	Sink         *sink.Config          `yaml:"sink,omitempty" validate:"dive,required"`
}

func GetDefaultConfigRaw

func GetDefaultConfigRaw() (*Config, error)

func (*Config) DeepCopy

func (c *Config) DeepCopy() *Config

func (*Config) SetDefaults

func (c *Config) SetDefaults()

func (*Config) Validate

func (c *Config) Validate() error

type Epoch

type Epoch struct {
	PipelineName string
	ReloadCount  int
	StartTime    time.Time
}

func NewEpoch

func NewEpoch(pipelineName string) *Epoch

func (*Epoch) Equal

func (e *Epoch) Equal(ae *Epoch) bool

func (*Epoch) Increase

func (e *Epoch) Increase()

func (*Epoch) IsEmpty

func (e *Epoch) IsEmpty() bool

func (*Epoch) String

func (e *Epoch) String() string

type Factory

type Factory func(info Info) api.Component

type Info

type Info struct {
	Stop         bool // lazy stop signal
	PipelineName string
	SurviveChan  chan api.Batch
	Epoch        *Epoch
	R            *RegisterCenter
	SinkCount    int
	EventPool    *event.Pool
}

type Pipeline

type Pipeline struct {
	Running bool
	// contains filtered or unexported fields
}

func NewPipeline

func NewPipeline(pipelineConfig *Config) *Pipeline

func (*Pipeline) Start

func (p *Pipeline) Start() error

func (*Pipeline) Stop

func (p *Pipeline) Stop()

type RegisterCenter

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

func NewRegisterCenter

func NewRegisterCenter() *RegisterCenter

func (*RegisterCenter) LoadCodeCategoryComponents

func (r *RegisterCenter) LoadCodeCategoryComponents(category api.Category) map[string]api.Component

func (*RegisterCenter) LoadCodeComponents

func (r *RegisterCenter) LoadCodeComponents() map[string]api.Component

func (*RegisterCenter) LoadCodeInterceptors

func (r *RegisterCenter) LoadCodeInterceptors() map[string]api.Interceptor

func (*RegisterCenter) LoadDefaultQueue

func (r *RegisterCenter) LoadDefaultQueue() api.Queue

func (*RegisterCenter) LoadInterceptors

func (r *RegisterCenter) LoadInterceptors() []api.Interceptor

func (*RegisterCenter) LoadQueue

func (r *RegisterCenter) LoadQueue(typename api.Type, name string) api.Queue

func (*RegisterCenter) LoadQueueListeners

func (r *RegisterCenter) LoadQueueListeners() []spi.QueueListener

func (*RegisterCenter) LoadSink

func (r *RegisterCenter) LoadSink(typename api.Type, name string) api.Sink

func (*RegisterCenter) LoadSource

func (r *RegisterCenter) LoadSource(typename api.Type, name string) api.Source

func (*RegisterCenter) LoadWithType

func (r *RegisterCenter) LoadWithType(typename api.Type, name string, componentType api.Category) api.Component

func (*RegisterCenter) Register

func (r *RegisterCenter) Register(component api.Component, name string) error

func (*RegisterCenter) RegisterListener

func (r *RegisterCenter) RegisterListener(listener spi.ComponentListener)

func (*RegisterCenter) RemoveByCode

func (r *RegisterCenter) RemoveByCode(code string)