rivo

package module

Go to main page

Versions in this module

v0

v0.4.1

Mar 6, 2025

v0.4.0

Mar 6, 2025

Changes in this version

+ func SegregateErrors[T any](p Pipeline[None, T]) (Pipeline[None, T], Pipeline[None, T])

+ type BatchOption func(*batchOptions) error

+ func BatchBufferSize(n int) BatchOption

+ func BatchMaxWait(d time.Duration) BatchOption

+ type DoOption func(*doOptions) error

+ func DoPoolSize(n int) DoOption

+ type ErrorStream = Stream[struct{}]

+ type FilterOption func(*filterOptions) error

+ func FilterBufferSize(n int) FilterOption

+ func FilterPoolSize(n int) FilterOption

+ type FromFuncOption func(*fromFuncOptions) error

+ func FromFuncBufferSize(bufferSize int) FromFuncOption

+ func FromFuncOnBeforeClose(f func(context.Context) error) FromFuncOption

+ func FromFuncPoolSize(poolSize int) FromFuncOption

+ type MapOption func(*mapOptions) error

+ func MapBufferSize(bufferSize int) MapOption

+ func MapPoolSize(poolSize int) MapOption

v0.3.1

Jan 13, 2025

v0.3.0

Jan 13, 2025

Changes in this version

+ type Pipeline func(ctx context.Context, stream Stream[T]) Stream[U]

v0.2.0

Jan 8, 2025

Changes in this version

type Pipeable

+ func Flatten[T any](opt ...Option) Pipeable[[]T, T]

v0.1.0

Jan 7, 2025

Changes in this version

type Pipeable

+ func Batch[T any](n int, maxWait time.Duration, opt ...Option) Pipeable[T, []T]

v0.0.0

Jan 5, 2025

Changes in this version

+ var ErrEOS = errors.New("end of stream")

+ func Segregate[T, U any](p Pipeable[T, U], predicate func(ctx context.Context, item Item[U]) bool) func(context.Context, Stream[T]) (Generator[U], Generator[U])

+ func Tee[T any](ctx context.Context, in Stream[T]) (Stream[T], Stream[T])

+ type DoFunc = func(context.Context, Item[T])

+ type FilterFunc = func(context.Context, Item[T]) (bool, error)

+ type ForEachFunc = func(context.Context, Item[T]) error

+ type FromFuncFunc = func(context.Context) (T, error)

+ type FromSeq2Value struct

+ Val1 T

+ Val2 U

+ type Generator = Pipeable[None, T]

+ func FromFunc[T any](f FromFuncFunc[T], options ...Option) Generator[T]

+ func FromSeq2[T, U any](seq iter.Seq2[T, U], opts ...Option) Generator[FromSeq2Value[T, U]]

+ func FromSeq[T any](seq iter.Seq[T], opt ...Option) Generator[T]

+ func Of[T any](items ...T) Generator[T]

+ type Item struct

+ Err error

+ Val T

+ func CollectWithContext[T any](ctx context.Context, in Stream[T]) []Item[T]

+ func Collect[T any](in Stream[T]) []Item[T]

+ type MapFunc = func(context.Context, Item[T]) (U, error)

+ type None struct

+ type Option func(*options)

+ func WithBufferSize(size int) Option

+ func WithOnBeforeClose(fn func(context.Context) error) Option

+ func WithPoolSize(size int) Option

+ func WithStopOnError(stop bool) Option

+ type Pipeable func(ctx context.Context, stream Stream[T]) Stream[U]

+ func Pipe2[A, B, C any](a Pipeable[A, B], b Pipeable[B, C]) Pipeable[A, C]

+ func Pipe3[A, B, C, D any](a Pipeable[A, B], b Pipeable[B, C], c Pipeable[C, D]) Pipeable[A, D]

+ func Pipe4[A, B, C, D, E any](a Pipeable[A, B], b Pipeable[B, C], c Pipeable[C, D], d Pipeable[D, E]) Pipeable[A, E]

+ func Pipe5[A, B, C, D, E, F any](a Pipeable[A, B], b Pipeable[B, C], c Pipeable[C, D], d Pipeable[D, E], ...) Pipeable[A, F]

+ func Pipe[A, B, C any](a Pipeable[A, B], b Pipeable[B, C]) Pipeable[A, C]

+ type Stream <-chan Item[T]

+ func OrDone[T any](ctx context.Context, in Stream[T]) Stream[T]

+ func TeeN[T any](ctx context.Context, in Stream[T], n int) []Stream[T]

+ type Sync = Pipeable[T, None]

+ func Connect[A any](pipeables ...Sync[A]) Sync[A]

+ func Do[T any](f DoFunc[T], opt ...Option) Sync[T]

+ type Transformer = Pipeable[T, U]

+ func Filter[T any](f FilterFunc[T], opt ...Option) Transformer[T, T]

+ func ForEach[T any](f ForEachFunc[T], opt ...Option) Transformer[T, struct{}]

+ func Map[T, U any](f MapFunc[T, U], opt ...Option) Transformer[T, U]

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL