pipeline

Documentation

Index

• Constants
• type Cfg
  • func NewCfg() *Cfg
  • func (c *Cfg) Add(role rw.Role, typ rw.Type, cfg any, extraCfg *rw.ExtraCfg) *Cfg
  • func (c *Cfg) AddCfg(cfg *rw.Cfg) *Cfg
• type Pipeline
  • func New() *Pipeline
  • func (p *Pipeline) GetCfg() any
  • func (p *Pipeline) Init() error
  • func (p *Pipeline) RWGroups() []*RWGroup
  • func (p *Pipeline) Result() *Result
  • func (p *Pipeline) Start() error
  • func (p *Pipeline) Stop() error
  • func (p *Pipeline) Type() any
• type RWGroup
  • func NewRWGroup() *RWGroup
  • func (g *RWGroup) FirstReader() rw.RW
  • func (g *RWGroup) Init() error
  • func (g *RWGroup) LastWriter() rw.RW
  • func (g *RWGroup) Readers() []rw.RW
  • func (g *RWGroup) Size() int
  • func (g *RWGroup) Start() error
  • func (g *RWGroup) Starter() rw.RW
  • func (g *RWGroup) Stop() error
  • func (g *RWGroup) Writers() []rw.RW
• type RWGroupCfg
  • func NewRWGroupCfg() *RWGroupCfg
  • func (c *RWGroupCfg) FromReader(typ rw.Type, cfg any, commonCfg *rw.ExtraCfg) *RWGroupCfg
  • func (c *RWGroupCfg) SetStarter(typ rw.Type, cfg any, commonCfg *rw.ExtraCfg) *RWGroupCfg
  • func (c *RWGroupCfg) ToWriter(typ rw.Type, cfg any, commonCfg *rw.ExtraCfg) *RWGroupCfg
• type Result

Constants

const PluginTypePipeline plugin.Type = "ppl"

const PluginTypeRWGroup plugin.Type = "rwg"

Types

type Cfg

type Cfg struct {
	RWs []*rw.Cfg `json:"rws" validate:"required" yaml:"rws"`
}

func NewCfg

func NewCfg() *Cfg

func (*Cfg) Add

func (c *Cfg) Add(role rw.Role, typ rw.Type, cfg any, extraCfg *rw.ExtraCfg) *Cfg

func (*Cfg) AddCfg

func (c *Cfg) AddCfg(cfg *rw.Cfg) *Cfg

type Pipeline

type Pipeline struct {
	runner.Runner
	*Cfg
	// contains filtered or unexported fields
}

func New

func New() *Pipeline

func (*Pipeline) GetCfg

func (p *Pipeline) GetCfg() any

func (*Pipeline) Init

func (p *Pipeline) Init() error

func (*Pipeline) RWGroups

func (p *Pipeline) RWGroups() []*RWGroup

func (*Pipeline) Result

func (p *Pipeline) Result() *Result

func (*Pipeline) Start

func (p *Pipeline) Start() error

func (*Pipeline) Stop

func (p *Pipeline) Stop() error

func (*Pipeline) Type

func (p *Pipeline) Type() any

type RWGroup

type RWGroup struct {
	runner.Runner
	*RWGroupCfg
	// contains filtered or unexported fields
}

func NewRWGroup

func NewRWGroup() *RWGroup

func (*RWGroup) FirstReader

func (g *RWGroup) FirstReader() rw.RW

func (*RWGroup) Init

func (g *RWGroup) Init() error

func (*RWGroup) LastWriter

func (g *RWGroup) LastWriter() rw.RW

func (*RWGroup) Readers

func (g *RWGroup) Readers() []rw.RW

func (*RWGroup) Size

func (g *RWGroup) Size() int

func (*RWGroup) Start

func (g *RWGroup) Start() error

func (*RWGroup) Starter

func (g *RWGroup) Starter() rw.RW

func (*RWGroup) Stop

func (g *RWGroup) Stop() error

func (*RWGroup) Writers

func (g *RWGroup) Writers() []rw.RW

type RWGroupCfg

type RWGroupCfg struct {
	Readers []*rw.Cfg `json:"readers" yaml:"readers"`
	Starter *rw.Cfg   `json:"starter" yaml:"starter"`
	Writers []*rw.Cfg `json:"writers" yaml:"writers"`
}

func NewRWGroupCfg

func NewRWGroupCfg() *RWGroupCfg

func (*RWGroupCfg) FromReader

func (c *RWGroupCfg) FromReader(typ rw.Type, cfg any, commonCfg *rw.ExtraCfg) *RWGroupCfg

func (*RWGroupCfg) SetStarter

func (c *RWGroupCfg) SetStarter(typ rw.Type, cfg any, commonCfg *rw.ExtraCfg) *RWGroupCfg

func (*RWGroupCfg) ToWriter

func (c *RWGroupCfg) ToWriter(typ rw.Type, cfg any, commonCfg *rw.ExtraCfg) *RWGroupCfg

type Result

type Result struct {
	Cfg     *Cfg             `json:"cfg"     yaml:"cfg"`
	Data    map[string]any   `json:"data"    yaml:"data"`
	RWsData []map[string]any `json:"rwsData" yaml:"rwsData"`
}