pipeline

Documentation

Index

• func Aggregate[K comparable, I1, I2, A any](ctx context.Context, in1 Aggregator[I1, K], in2 Aggregator[I2, K], ...) (chan A, chan error)
• func Aggregate3[K comparable, I1, I2, I3, A any](ctx context.Context, in1 Aggregator[I1, K], in2 Aggregator[I2, K], ...) (chan A, chan error)
• func Chan[I any](f func() (I, error)) (chan I, chan error)
• func Merge[T any](ctx context.Context, chs ...chan T) <-chan T
• func PassingFn[In any](in In) (In, error)
• 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 StaticKeyFn[T any](i T) int
• func Step[In, Out any](ctx context.Context, chIn <-chan In, ...) (chan Out, chan error)
• func TestAggregate3(t *testing.T)
• func TestAggregate3Error(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, error)) (chan A, chan error)

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, error)) (chan A, chan error)

func Chan

func Chan[I any](f func() (I, error)) (chan I, 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 PassingFn

func PassingFn[In any](in In) (In, error)

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 StaticKeyFn

func StaticKeyFn[T any](i T) int

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)

func TestAggregate3Error

func TestAggregate3Error(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