pipeline

type DataProcessor ¶ added in v0.2.0

type DataProcessor[T any] interface {
	Process(T) (T, error)
}

DataProcessor is an interface for processing data.

type DataWriter ¶

type DataWriter[T any] interface {
	Write(T) (n int, err error)
	Close() (err error)
}

DataWriter is an interface for writing data to a destination.

type Mode ¶ added in v0.4.0

type Mode int

const (
	// ModeSingle default pipeline mode,
	// when all workers from each stage linked to the other stage workers with one channel.
	ModeSingle Mode = iota
	// ModeParallel advanced pipeline mode,
	// when each worker in one stage linked to each worker f another stage with one channel.
	ModeParallel
	// ModeSingleParallel advanced pipeline mode,
	// When first stage with second stage is linked in single mode, and second and last stages linked in parallel.
	// SplitFunction must be set, for using this mode.
	ModeSingleParallel
)

type Pipeline ¶

type Pipeline[T any] struct {
	// contains filtered or unexported fields
}

Pipeline runs a series of workers in parallel. All workers run at the same time in separate goroutines. Each stage of workers are connected by a single channel that is the size of the number of workers in the stage. The pipeline stops when a worker returns an error or all workers are done. workers should not close the send channel, the pipeline stages handle that. Pipelines can be chained together by using them as workers.

func NewPipeline[T any](mode Mode, sf splitFunc[T], workGroups ...[]Worker[T]) (*Pipeline[T], error)

func (dp *Pipeline[T]) GetMetrics() (in, out int)

func (dp *Pipeline[T]) Run(ctx context.Context) error

type ProcessorWorker ¶ added in v0.2.0

type ProcessorWorker[T any] struct {
	// contains filtered or unexported fields
}

ProcessorWorker implements the pipeline.Worker interface. It wraps a DataProcessor and processes data with it.

func NewProcessorWorker[T any](processor DataProcessor[T]) *ProcessorWorker[T]

func (w *ProcessorWorker[T]) GetMetrics() (in, out int)

func (w *ProcessorWorker[T]) Run(ctx context.Context) error

func (w *ProcessorWorker[T]) SetReceiveChan(c <-chan T)

func (w *ProcessorWorker[T]) SetSendChan(c chan<- T)

type Route ¶ added in v0.4.0

type Route[T any] struct {
	// contains filtered or unexported fields
}

Route describes how workers will communicate through stages. Each stage must have the corresponding route. The Route consists of input and output rule.

func NewParallelRoutes[T any](stagesNum int) []Route[T]

func NewSingleParallelRoutes[T any](sf splitFunc[T]) []Route[T]

func NewSingleRoutes[T any](stagesNum int) []Route[T]

type RouteRule ¶ added in v0.4.0

type RouteRule[T any] struct {
	// contains filtered or unexported fields
}

RouteRule describes how exactly stages will communicate with each other.

func NewRouteRuleParallel[T any](bufferSize int, sf splitFunc[T]) *RouteRule[T]

func NewRouteRuleSingle[T any](bufferSize int, sf splitFunc[T]) *RouteRule[T]

type Worker ¶

type Worker[T any] interface {
	SetSendChan(chan<- T)
	SetReceiveChan(<-chan T)
	Run(context.Context) error
	GetMetrics() (int, int)
}

Worker is an interface for a pipeline item. Each worker has send and receive channels that connect it to the previous and next stage in the pipeline. The Run method starts the worker.

func NewReadWorker[T any](reader dataReader[T]) Worker[T]

func NewWriteWorker[T any](writer DataWriter[T], limiter *rate.Limiter) Worker[T]