chunkflow

package module

Go to main page

Versions in this module

v0

v0.1.0

Sep 20, 2026

Changes in this version

type Builder

+ func (b Builder) SeqErr[T any](seq iter.Seq2[T, error]) Stream[T]

+ type Entry struct

+ Key K

+ Value V

type Stream

+ func (s Stream[T]) ChunkTimeout[R []T](size int, maxWait time.Duration) Stream[R]

+ func (s Stream[T]) ZipCtx[O, R any](other Stream[O], fn func(ctx context.Context, a T, b O) (R, error)) Stream[R]

+ func (s Stream[T]) Zip[O, R any](other Stream[O], fn func(a T, b O) R) Stream[R]

v0.0.3

Sep 19, 2026

Changes in this version

+ var ErrEmpty = errors.New("empty stream")

+ func Suppress(err error) error

type StepOption

+ func WithUnordered() StepOption

type Stream

+ func (s Stream[T]) Drain() error

+ func (s Stream[T]) ReduceByCtx[K comparable, R any](key func(T) K, init R, fn func(ctx context.Context, acc R, item T) (R, error)) (map[K]R, error)

+ func (s Stream[T]) ReduceBy[K comparable, R any](key func(T) K, init R, fn func(acc R, item T) R) (map[K]R, error)

+ func (s Stream[T]) Transform[R any](transform func(iter.Seq2[T, error]) iter.Seq2[R, error]) Stream[R]

v0.0.2

Sep 16, 2026

Changes in this version

+ var ErrPanic = errors.New("panic in callback")

+ type Builder struct

+ func New(ctx context.Context, opts ...Option) Builder

+ func (b Builder) Chan[T any](ch <-chan T) Stream[T]

+ func (b Builder) Seq2[T any](seq iter.Seq2[T, error]) Stream[T]

+ func (b Builder) Seq[T any](seq iter.Seq[T]) Stream[T]

type Option

+ func WithOnError(fn func(error)) Option

+ type StepOption func(*options)

type Stream

+ func Compact[T comparable](stream Stream[T]) Stream[T]

+ func Concat[T any](streams ...Stream[T]) Stream[T]

+ func Merge[T any](streams ...Stream[T]) Stream[T]

+ func (s Stream[T]) AllCtx(predicate func(ctx context.Context, item T) (bool, error)) (bool, error)

+ func (s Stream[T]) AnyCtx(predicate func(ctx context.Context, item T) (bool, error)) (bool, error)

+ func (s Stream[T]) CircuitBreaker(maxConsecutiveFailures int) Stream[T]

+ func (s Stream[T]) CompactFunc(eq func(a, b T) bool) Stream[T]

+ func (s Stream[T]) FilterCtx(predicate func(ctx context.Context, item T) (bool, error), opts ...StepOption) Stream[T]

+ func (s Stream[T]) ForEachCtx(fn func(ctx context.Context, item T) error) error

+ func (s Stream[T]) MapCtx[R any](mapFn func(ctx context.Context, item T) (R, error), opts ...StepOption) Stream[R]

+ func (s Stream[T]) Opts(opts ...Option) Stream[T]

+ func (s Stream[T]) ReduceCtx[R any](init R, fn func(ctx context.Context, acc R, item T) (R, error)) (R, error)

+ func (s Stream[T]) SkipWhile(predicate func(T) bool) Stream[T]

+ func (s Stream[T]) SkipWhileCtx(predicate func(ctx context.Context, item T) (bool, error)) Stream[T]

+ func (s Stream[T]) TakeWhile(predicate func(T) bool) Stream[T]

+ func (s Stream[T]) TakeWhileCtx(predicate func(ctx context.Context, item T) (bool, error)) Stream[T]

+ func (s Stream[T]) Tap(fn func(T)) Stream[T]

+ func (s Stream[T]) TapCtx(fn func(ctx context.Context, item T) error, opts ...StepOption) Stream[T]

v0.0.1

Sep 15, 2026

Changes in this version

+ var ErrSuppressed = errors.New("suppressed by circuit breaker")

+ type IoStream struct

+ func IoFlatten[E any](stream IoStream[[]E]) IoStream[E]

+ func NewIoStream2[T any](ctx context.Context, seq iter.Seq2[T, error]) IoStream[T]

+ func NewIoStream[T any](ctx context.Context, seq iter.Seq[T]) IoStream[T]

+ func (s IoStream[T]) All(predicate func(T) bool) (bool, error)

+ func (s IoStream[T]) AllAsync(predicate func(ctx context.Context, item T) (bool, error)) (bool, error)

+ func (s IoStream[T]) Any(predicate func(T) bool) (bool, error)

+ func (s IoStream[T]) AnyAsync(predicate func(ctx context.Context, item T) (bool, error)) (bool, error)

+ func (s IoStream[T]) Chunk[R []T](size int) IoStream[R]

+ func (s IoStream[T]) CircuitBreaker(maxConsecutiveFailures int) IoStream[T]

+ func (s IoStream[T]) Collect() ([]T, error)

+ func (s IoStream[T]) Count() (int, error)

+ func (s IoStream[T]) Exec() error

+ func (s IoStream[T]) Filter(predicate func(T) bool) IoStream[T]

+ func (s IoStream[T]) FilterAsync(predicate func(ctx context.Context, item T) (bool, error), opts ...Option) IoStream[T]

+ func (s IoStream[T]) First() (T, bool, error)

+ func (s IoStream[T]) ForEach(fn func(T)) error

+ func (s IoStream[T]) ForEachAsync(fn func(ctx context.Context, item T) error) error

+ func (s IoStream[T]) Last() (T, bool, error)

+ func (s IoStream[T]) MapAsync[R any](mapFn func(ctx context.Context, item T) (R, error), opts ...Option) IoStream[R]

+ func (s IoStream[T]) Map[R any](mapFn func(T) R) IoStream[R]

+ func (s IoStream[T]) Opts(opts ...Option) IoStream[T]

+ func (s IoStream[T]) Reduce(init T, fn func(item, acc T) T) (T, error)

+ func (s IoStream[T]) ReduceAsync(init T, fn func(ctx context.Context, item, acc T) (T, error), opts ...Option) (T, error)

+ func (s IoStream[T]) Seq() iter.Seq2[T, error]

+ func (s IoStream[T]) Skip(nr int) IoStream[T]

+ func (s IoStream[T]) Take(nr int) IoStream[T]

+ func (s IoStream[T]) Through[R any](transform func(IoStream[T]) IoStream[R]) IoStream[R]

+ type Option func(*options)

+ func WithDiscardLogger() Option

+ func WithLogger(logger *slog.Logger) Option

+ func WithParallel(concurrency int) Option

+ type Stream struct

+ func Flatten[E any](stream Stream[[]E]) Stream[E]

+ func NewStream[T any](seq iter.Seq[T]) Stream[T]

+ func (a Stream[T]) All(predicate func(T) bool) bool

+ func (a Stream[T]) Any(predicate func(T) bool) bool

+ func (a Stream[T]) Chunk[R []T](size int) Stream[R]

+ func (a Stream[T]) Collect() []T

+ func (a Stream[T]) Count() int

+ func (a Stream[T]) Exec()

+ func (a Stream[T]) Filter(predicate func(T) bool) Stream[T]

+ func (a Stream[T]) First() (T, bool)

+ func (a Stream[T]) ForEach(fn func(T))

+ func (a Stream[T]) Last() (T, bool)

+ func (a Stream[T]) Map[R any](mapFn func(T) R) Stream[R]

+ func (a Stream[T]) Reduce(init T, fn func(item, acc T) T) T

+ func (a Stream[T]) Seq() iter.Seq[T]

+ func (a Stream[T]) Skip(nr int) Stream[T]

+ func (a Stream[T]) Take(nr int) Stream[T]

+ func (a Stream[T]) Through[R any](transform func(Stream[T]) Stream[R]) Stream[R]

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