stages

Documentation

Index

• type AggregateStage
  • func (s AggregateStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan []T
• type BatchStage
  • func (s BatchStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan []T
• type ExpandSliceStage
  • func (s ExpandSliceStage[T]) Create(ctx pipeline.Context, in <-chan []T) <-chan T
• type ExpandStage
  • func (s ExpandStage[In, Out]) Create(ctx pipeline.Context, in <-chan In) <-chan Out
• type Expander
• type FanInStage
  • func (s FanInStage[T]) Create(ctx pipeline.Context, in pipeline.MultiChannelReceiver[T]) <-chan T
• type FanOutRoundRobinStage
  • func (s FanOutRoundRobinStage[T]) Create(ctx pipeline.Context, in <-chan T) pipeline.MultiChannelReceiver[T]
• type FanOutStage
  • func (s FanOutStage[T]) Create(ctx pipeline.Context, in <-chan T) pipeline.MultiChannelReceiver[T]
• type Filter
• type FilterStage
  • func (s FilterStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan T
• type FlattenStage
  • func (s FlattenStage[T]) Create(ctx pipeline.Context, in <-chan []T) <-chan T
• type LimitStage
  • func (s LimitStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan T
• type ParallelTransformStage
  • func (s ParallelTransformStage[In, Out]) Create(ctx pipeline.Context, in <-chan In) <-chan Out
• type ReduceStage
  • func (s ReduceStage[T, Acc]) Create(ctx pipeline.Context, in <-chan T) <-chan Acc
• type Reducer
• type Selector
• type SplitStage
  • func (s SplitStage[T]) Create(ctx pipeline.Context, in <-chan T) pipeline.MultiChannelReceiver[T]
• type TransformStage
  • func (s TransformStage[In, Out]) Create(ctx pipeline.Context, in <-chan In) <-chan Out
• type Transformer

Types

type AggregateStage

type AggregateStage[T any] struct {
	Name   string
	Buffer int
}

func (AggregateStage[T]) Create

func (s AggregateStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan []T

type BatchStage

type BatchStage[T any] struct {
	Name      string
	Buffer    int
	BatchSize int
}

func (BatchStage[T]) Create

func (s BatchStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan []T

type ExpandSliceStage

type ExpandSliceStage[T any] struct {
	Name   string
	Buffer int
}

func (ExpandSliceStage[T]) Create

func (s ExpandSliceStage[T]) Create(ctx pipeline.Context, in <-chan []T) <-chan T

type ExpandStage

type ExpandStage[In any, Out any] struct {
	Name     string
	Buffer   int
	Expander Expander[In, Out]
}

func (ExpandStage[In, Out]) Create

func (s ExpandStage[In, Out]) Create(ctx pipeline.Context, in <-chan In) <-chan Out

type Expander

type Expander[In any, Out any] func(context.Context, In) iter.Seq2[Out, error]

type FanInStage

type FanInStage[T any] struct {
	Name   string
	Buffer int
}

func (FanInStage[T]) Create

func (s FanInStage[T]) Create(ctx pipeline.Context, in pipeline.MultiChannelReceiver[T]) <-chan T

type FanOutRoundRobinStage

type FanOutRoundRobinStage[T any] struct {
	Name        string
	OutputCount int
	Buffer      int
}

func (FanOutRoundRobinStage[T]) Create

func (s FanOutRoundRobinStage[T]) Create(ctx pipeline.Context, in <-chan T) pipeline.MultiChannelReceiver[T]

type FanOutStage

type FanOutStage[T any] struct {
	Name        string
	OutputCount int
	Buffer      int
}

func (FanOutStage[T]) Create

func (s FanOutStage[T]) Create(ctx pipeline.Context, in <-chan T) pipeline.MultiChannelReceiver[T]

type Filter

type Filter[T any] func(context.Context, T) (bool, error)

type FilterStage

type FilterStage[T any] struct {
	Name   string
	Buffer int
	Filter Filter[T]
}

func (FilterStage[T]) Create

func (s FilterStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan T

type FlattenStage

type FlattenStage[T any] struct {
	Name   string
	Buffer int
}

func (FlattenStage[T]) Create

func (s FlattenStage[T]) Create(ctx pipeline.Context, in <-chan []T) <-chan T

type LimitStage

type LimitStage[T any] struct {
	Name   string
	Buffer int
	Limit  int
}

func (LimitStage[T]) Create

func (s LimitStage[T]) Create(ctx pipeline.Context, in <-chan T) <-chan T

type ParallelTransformStage

type ParallelTransformStage[In any, Out any] struct {
	Name        string
	Buffer      int
	Workers     int
	Transformer Transformer[In, Out]
}

func (ParallelTransformStage[In, Out]) Create

func (s ParallelTransformStage[In, Out]) Create(ctx pipeline.Context, in <-chan In) <-chan Out

type ReduceStage

type ReduceStage[T any, Acc any] struct {
	Name    string
	Buffer  int
	Initial Acc
	Reducer Reducer[T, Acc]
}

func (ReduceStage[T, Acc]) Create

func (s ReduceStage[T, Acc]) Create(ctx pipeline.Context, in <-chan T) <-chan Acc

type Reducer

type Reducer[T any, Acc any] func(context.Context, Acc, T) (Acc, error)

type Selector

type Selector[T any] func(context.Context, T) int

type SplitStage

type SplitStage[T any] struct {
	Name        string
	OutputCount int
	Buffer      int
	Selector    Selector[T]
}

func (SplitStage[T]) Create

func (s SplitStage[T]) Create(ctx pipeline.Context, in <-chan T) pipeline.MultiChannelReceiver[T]

type TransformStage

type TransformStage[In any, Out any] struct {
	Name        string
	Buffer      int
	Transformer Transformer[In, Out]
}

func (TransformStage[In, Out]) Create

func (s TransformStage[In, Out]) Create(ctx pipeline.Context, in <-chan In) <-chan Out

type Transformer

type Transformer[In any, Out any] func(context.Context, In) (*Out, error)