pipe

Documentation

Index

• func DiscardFunc[V any](in <-chan V) func()
• func FirstFunc[V any](in <-chan V, fn func(V)) func()
• func GroupBy[K comparable, V any](ctx context.Context, in <-chan *KV[K, V]) <-chan *KV[K, []V]
• func LastFunc[V any](in <-chan V, fn func(V)) func()
• func Merge[t any](ctx context.Context, channels ...<-chan t) <-chan t
• func ParDo[inType any, outType any](ctx context.Context, parallelism int, in <-chan inType, f func(inType) outType) <-chan outType
• func ParDoErr[inType any, outType any, chanInType readable[inType]](ctx context.Context, parallelism int, in chanInType, ...) (<-chan outType, <-chan error)
• func ParDoFilter[inType any](ctx context.Context, parallelism int, in <-chan inType, f func(inType) bool) <-chan inType
• func ParDoFilterErr[inType any](ctx context.Context, parallelism int, in <-chan inType, ...) (<-chan inType, <-chan error)
• func ToArrayFunc[V any](in <-chan V, out *[]V) func()
• func ToMapFunc[K comparable, V any](in <-chan *KV[K, V], out *map[K]V) func()
• func TryWrite[chanType any](ctx context.Context, out chan<- chanType, val chanType) bool
• func Wait(funcs ...func())
• type KV

Functions

func DiscardFunc

func DiscardFunc[V any](in <-chan V) func()

The Discard function returns a function which, when called, consumes all values from the input channel. The returned function only terminates when the input channel is closed.

func FirstFunc

func FirstFunc[V any](in <-chan V, fn func(V)) func()

The First function returns a function which, when called, reads the first value from the input channel, and passes that value to the given function. If the channel never receives a value, then the given function is never called. The returned function consumes all values from the input channel, and only terminates when the input channel is closed.

func GroupBy

func GroupBy[K comparable, V any](ctx context.Context, in <-chan *KV[K, V]) <-chan *KV[K, []V]

The GroupBy function reads from the input channel and groups the values by key. This requires storing all of the values in memory, so it is not suitable for large datasets.

func LastFunc

func LastFunc[V any](in <-chan V, fn func(V)) func()

The Last function returns a function which, when called, reads all values from the input channel, and passes the last value to the given function. If the channel never receives a value, then the given function is never called. The returned function only terminates when the input channel is closed.

func Merge

func Merge[t any](ctx context.Context, channels ...<-chan t) <-chan t

Merge() merges multiple channels into a single channel.

func ParDo

func ParDo[inType any, outType any](ctx context.Context, parallelism int, in <-chan inType, f func(inType) outType) <-chan outType

ParDo() runs a function in parallel on each value from the input channel.

func ParDoErr

func ParDoErr[inType any, outType any, chanInType readable[inType]](ctx context.Context, parallelism int, in chanInType, f func(inType) (outType, error)) (<-chan outType, <-chan error)

func ParDoFilter

func ParDoFilter[inType any](ctx context.Context, parallelism int, in <-chan inType, f func(inType) bool) <-chan inType

func ParDoFilterErr

func ParDoFilterErr[inType any](ctx context.Context, parallelism int, in <-chan inType, f func(inType) (bool, error)) (<-chan inType, <-chan error)

func ToArrayFunc

func ToArrayFunc[V any](in <-chan V, out *[]V) func()

The ToArray function returns a function which, when called, reads from the input channel and appends the values to the output slice. The returned function terminates when the input channel is closed.

func ToMapFunc

func ToMapFunc[K comparable, V any](in <-chan *KV[K, V], out *map[K]V) func()

The ToMap function returns a function which, when called, consumes all key-value pairs from the input channel and stores them in the given map. The returned function only terminates when the input channel is closed. If the same key is encountered multiple times, the value in the map will be the last value seen.

func TryWrite

func TryWrite[chanType any](ctx context.Context, out chan<- chanType, val chanType) bool

TryWrite() blocks until either the value is written to the channel or the context becomes done.

Returns true if the value was written, false if the context was done.

func Wait

func Wait(funcs ...func())

Wait runs the given functions in new goroutines and waits for all of them to finish.

Types

type KV

type KV[K comparable, V any] struct {
	Key K
	Val V
}

KV is a simple key-value pair.