pipe

Oct 1, 2023

Changes in this version

type MapOp

+ func (op *MapOp[S, T]) ConcurrentPreservingOrder(concurrency int, opts ...task.Option) Pipe[S, T]

Aug 3, 2023

Changes in this version

+ type FlattenSliceOp struct

+ func FlattenSlice() *FlattenSliceOp[S]

+ func (op *FlattenSliceOp) AsPipe(opts ...task.Option) Pipe[S, T]

+ func (op *FlattenSliceOp) AsTask(opts ...task.Option) task.Task[S, T]

+ type SelectOp struct

+ func Select(predicate func(S) bool) *SelectOp[S]

+ func (op *SelectOp) AsPipe(opts ...task.Option) Pipe[S, T]

+ func (op *SelectOp) AsTask(opts ...task.Option) task.Task[S, T]

+ type TakeOp struct

+ func Take(n int) *TakeOp[S]

+ func (op *TakeOp) AsPipe(opts ...task.Option) Pipe[S, T]

+ func (op *TakeOp) AsTask(opts ...task.Option) task.Task[S, T]

type TapOp

+ func (op *TapOp) Concurrent(concurrency int, opts ...task.Option) Pipe[S, T]

Aug 2, 2023

Changes in this version

+ type BatchOp struct

+ func Batch(size int) *BatchOp[S]

+ func (op *BatchOp[S]) AsPipe(opts ...task.Option) Pipe[S, []S]

+ func (op *BatchOp[S]) AsTask() task.Task[S, []S]

type FanoutOp

+ func (op *FanoutOp[S, I, T]) AsPipe(opts ...task.Option) Pipe[S, T]

type MapOp

+ func (op *MapOp[S, T]) AsPipe(opts ...task.Option) Pipe[S, T]

type TapOp

+ func (op *TapOp[S]) AsPipe(opts ...task.Option) Pipe[S, S]

Jul 25, 2023

Changes in this version

+ type FanoutAggregateFn func(context.Context, []I, chan<- T) error

+ type FanoutMapFn func(context.Context, []I) (T, error)

+ type FanoutOp struct

+ func Fanout(aggFn FanoutAggregateFn[I, T]) *FanoutOp[S, I, T]

+ func FanoutWithMap(mapFn FanoutMapFn[I, T]) *FanoutOp[S, I, T]

+ func (op *FanoutOp[S, I, T]) Add(t task.Task[S, I], inBuffer, outBuffer int)

+ func (op *FanoutOp[S, I, T]) AsTask() task.Task[S, T]

+ func (op *FanoutOp[S, I, T]) Run(ctx context.Context, in <-chan S, out chan<- T) error

+ type MapFn func(context.Context, S) (T, error)

+ type MapOp struct

+ func Map(fn MapFn[S, T]) *MapOp[S, T]

+ func MapWithCache(fn MapFn[S, T], sp cache.Spec[S, T, K, V]) *MapOp[S, T]

+ func (op *MapOp[S, T]) AsTask() task.Task[S, T]

+ func (op *MapOp[S, T]) Concurrent(concurrency int) Pipe[S, T]

+ type Pipe task.Task[S, T]

+ func Concurrent(ps []Pipe[S, T]) Pipe[S, T]

+ func ConcurrentFromFn(fn PipeFn[S, T], concurrency int) Pipe[S, T]

+ func FromFn(fn PipeFn[S, T]) Pipe[S, T]

+ type PipeFn func(ctx context.Context, in <-chan S, out chan<- T) error

+ type TapFn func(context.Context, S) error

+ type TapOp struct

+ func Tap(fn TapFn[S]) *TapOp[S]

+ func (op *TapOp[S]) AsTask() task.Task[S, S]