pipeline

type Source ¶

type Source[T any] func(context.Context, *errgroup.Group) <-chan T

func NewSource[T any](sourceFn SourceFn[T]) Source[T]

type SourceFn ¶

type SourceFn[T any] func(ctx context.Context, put func(T) error) error

SourceFn is a function to produce work items. It should call put(item) to add an item into the pipeline for processing.

type Stage ¶

type Stage[T any] func(context.Context, *errgroup.Group, <-chan T) <-chan T

func NewStage[T any](fn StageFn[T]) Stage[T]

type StageFn ¶

type StageFn[T any] func(ctx context.Context, item T) (T, error)

StageFn is a function process a work item. It takes an item as an input an return a new one after processing. The last stage is a sink stage and its output will be ignored.