pipeline

Nov 25, 2022

Changes in this version

type Pipeline

+ func (p *Pipeline) DataItemCount() int

+ func (p *Pipeline) DecDataItemCount() int

+ func (p *Pipeline) IncDataItemCount() int

type StageParams

+ Pipeline func() *Pipeline

type TaskParams

+ Pipeline func() *Pipeline

Mar 10, 2022

Changes in this version

+ func SendData(ctx context.Context, stage string, data Data, tp TaskParams)

+ type Data interface

+ Clone func() Data

+ MarkAsProcessed func()

+ type InputSource interface

+ Data func() Data

+ Error func() error

+ Next func(context.Context) bool

+ type OutputSink interface

+ Consume func(context.Context, Data) error

+ type Pipeline struct

+ func NewPipeline(stages ...Stage) *Pipeline

+ func (p *Pipeline) Execute(ctx context.Context, src InputSource, sink OutputSink) error

+ func (p *Pipeline) ExecuteBuffered(ctx context.Context, src InputSource, sink OutputSink, bufsize int) error

+ type SinkFunc func(context.Context, Data) error

+ func (f SinkFunc) Consume(ctx context.Context, data Data) error

+ type Stage interface

+ ID func() string

+ Run func(context.Context, StageParams)

+ func Broadcast(id string, tasks ...Task) Stage

+ func DynamicPool(id string, task Task, max int) Stage

+ func FIFO(id string, task Task) Stage

+ func FixedPool(id string, task Task, num int) Stage

+ func Parallel(id string, tasks ...Task) Stage

+ type StageParams interface

+ DataQueue func() queue.Queue

+ Error func() queue.Queue

+ Input func() <-chan Data

+ Output func() chan<- Data

+ Position func() int

+ Registry func() StageRegistry

+ type StageRegistry map[string]queue.Queue

+ type Task interface

+ Process func(context.Context, Data, TaskParams) (Data, error)

+ type TaskFunc func(context.Context, Data, TaskParams) (Data, error)

+ func (f TaskFunc) Process(ctx context.Context, data Data, params TaskParams) (Data, error)

+ type TaskParams interface

+ Registry func() StageRegistry