pipeline

Documentation

Index

• func Aggregate[K comparable, I1, I2, A any](ctx context.Context, in1 Aggregator[I1, K], in2 Aggregator[I2, K], ...) chan A
• func Aggregate3[K comparable, I1, I2, I3, A any](ctx context.Context, in1 Aggregator[I1, K], in2 Aggregator[I2, K], ...) chan A
• func FlatMap[T any](ctx context.Context, in chan []T) chan T
• func Map[I, O any](ctx context.Context, in chan I, f func(I) (O, error)) (chan O, chan error)
• func Merge[T any](ctx context.Context, chs ...chan T) <-chan T
• func Source[In any](ctx context.Context, in ...In) chan In
• func Split[In, O1, O2 any](ctx context.Context, chIn <-chan In, sa1 *stepaction.StepAction[In, O1], ...) (chan O1, chan O2, chan error)
• func Step[In, Out any](ctx context.Context, chIn <-chan In, ...) (chan Out, chan error)
• func TestAggregate3(t *testing.T)
• type Aggregator
• type StepOption
  • func WithLimit(l int64) StepOption

Functions

func Aggregate

func Aggregate[K comparable, I1, I2, A any](ctx context.Context, in1 Aggregator[I1, K], in2 Aggregator[I2, K], aggFn func(K, I1, I2) A) chan A

func Aggregate3

func Aggregate3[K comparable, I1, I2, I3, A any](ctx context.Context, in1 Aggregator[I1, K], in2 Aggregator[I2, K], in3 Aggregator[I3, K], aggFn func(K, I1, I2, I3) A) chan A

func FlatMap

func FlatMap[T any](ctx context.Context, in chan []T) chan T

func Map

func Map[I, O any](ctx context.Context, in chan I, f func(I) (O, error)) (chan O, chan error)

func Merge

func Merge[T any](ctx context.Context, chs ...chan T) <-chan T

Merge collects data from all input channels and send it to output channel.

All input and output channels have same type.

func Source

func Source[In any](ctx context.Context, in ...In) chan In

func Split

func Split[In, O1, O2 any](
	ctx context.Context,
	chIn <-chan In,
	sa1 *stepaction.StepAction[In, O1],
	sa2 *stepaction.StepAction[In, O2],
	opts ...StepOption,
) (chan O1, chan O2, chan error)

func Step

func Step[In, Out any](
	ctx context.Context,
	chIn <-chan In,
	stepAction *stepaction.StepAction[In, Out],
	opts ...StepOption,
) (chan Out, chan error)

func TestAggregate3

func TestAggregate3(t *testing.T)

Types

type Aggregator

type Aggregator[I, K any] struct {
	Ch    chan I
	KeyFn func(I) K
}

type StepOption

type StepOption func(*stepConfig)

func WithLimit

func WithLimit(l int64) StepOption