pipeline

Documentation

Index

• type Payload
• type Pipeline
  • func New(stages ...StageRunner) *Pipeline
  • func (p *Pipeline) Process(ctx context.Context, source Source, sink Sink) error
• type Processor
• type ProcessorFunc
  • func (f ProcessorFunc) Process(ctx context.Context, p Payload) (Payload, error)
• type Sink
• type Source
• type StageParams
• type StageRunner
  • func Broadcast(procs ...Processor) StageRunner
  • func DynamicWorkerPool(proc Processor, maxWorkers int) StageRunner
  • func FIFO(proc Processor) StageRunner
  • func FixedWorkerPool(proc Processor, numWorkers int) StageRunner

Types

type Payload

type Payload interface {
	Clone() Payload
	MarkAsProcessed()
}

Payload is implemented by values that can be sent through a pipeline

type Pipeline

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

func New

func New(stages ...StageRunner) *Pipeline

func (*Pipeline) Process

func (p *Pipeline) Process(ctx context.Context, source Source, sink Sink) error

type Processor

type Processor interface {
	Process(context.Context, Payload) (Payload, error)
}

type ProcessorFunc

type ProcessorFunc func(context.Context, Payload) (Payload, error)

func (ProcessorFunc) Process

func (f ProcessorFunc) Process(ctx context.Context, p Payload) (Payload, error)

type Sink

type Sink interface {
	Consume(context.Context, Payload) error
}

type Source

type Source interface {
	Next(context.Context) bool
	Payload() Payload
	Error() error
}

type StageParams

type StageParams interface {
	StageIndex() int
	Input() <-chan Payload
	Output() chan<- Payload
	Error() chan<- error
}

type StageRunner

type StageRunner interface {
	Run(context.Context, StageParams)
}

func Broadcast

func Broadcast(procs ...Processor) StageRunner

func DynamicWorkerPool

func DynamicWorkerPool(proc Processor, maxWorkers int) StageRunner

func FIFO

func FIFO(proc Processor) StageRunner

func FixedWorkerPool

func FixedWorkerPool(proc Processor, numWorkers int) StageRunner