stream

Documentation

Overview

Package stream contains domain-neutral, typed lazy stream operators.

Index

• type StreamOp
  • func Batch[T any](size int) StreamOp[T, []T]
  • func BatchByTime[T any](ctx context.Context, size int, timeout time.Duration) StreamOp[T, []T]
  • func Collect[T any](maxItems int) StreamOp[T, []T]
  • func Debounce[T any](ctx context.Context, interval time.Duration) StreamOp[T, T]
  • func Filter[T any](pred func(T) bool) StreamOp[T, T]
  • func FlatMap[In, Out any](fn func(In) iter.Seq2[Out, error]) StreamOp[In, Out]
  • func Map[In, Out any](fn func(In) Out) StreamOp[In, Out]
  • func MapE[In, Out any](fn func(In) (Out, error)) StreamOp[In, Out]
  • func Reduce[T, Acc any](initial Acc, fn func(Acc, T) (Acc, error)) StreamOp[T, Acc]
  • func Take[T any](n int) StreamOp[T, T]
  • func Throttle[T any](interval time.Duration) StreamOp[T, T]
  • func Window[T any](size, step int) StreamOp[T, []T]
  • func WithContext[T any](ctx context.Context) StreamOp[T, T]

Types

type StreamOp

type StreamOp[In, Out any] func(iter.Seq2[In, error]) iter.Seq2[Out, error]

StreamOp transforms a lazy stream without erasing its item types.

func Batch

func Batch[T any](size int) StreamOp[T, []T]

Batch groups successful items into bounded batches. A final partial batch is emitted when the upstream ends normally.

func BatchByTime

func BatchByTime[T any](ctx context.Context, size int, timeout time.Duration) StreamOp[T, []T]

BatchByTime groups items up to a maximum size or until the timeout interval elapses.

func Collect

func Collect[T any](maxItems int) StreamOp[T, []T]

Collect materializes a bounded stream into one item. A non-positive limit is rejected at iteration time so the operator remains lazy.

func Debounce

func Debounce[T any](ctx context.Context, interval time.Duration) StreamOp[T, T]

Debounce drops items followed by another item within the interval.

func Filter

func Filter[T any](pred func(T) bool) StreamOp[T, T]

Filter forwards only items accepted by pred.

func FlatMap

func FlatMap[In, Out any](fn func(In) iter.Seq2[Out, error]) StreamOp[In, Out]

FlatMap expands each successful item sequentially. The inner stream must honor the downstream yield result; no goroutine is created per item.

func Map

func Map[In, Out any](fn func(In) Out) StreamOp[In, Out]

Map transforms every successful item. It does not intercept errors.

func MapE

func MapE[In, Out any](fn func(In) (Out, error)) StreamOp[In, Out]

MapE transforms every successful item and can stop with an error.

func Reduce

func Reduce[T, Acc any](initial Acc, fn func(Acc, T) (Acc, error)) StreamOp[T, Acc]

Reduce consumes the complete stream and emits one accumulated value.

func Take

func Take[T any](n int) StreamOp[T, T]

Take forwards at most n successful items. n <= 0 yields an empty stream.

func Throttle

func Throttle[T any](interval time.Duration) StreamOp[T, T]

Throttle drops items that arrive faster than the specified interval. Zero allocations.

func Window

func Window[T any](size, step int) StreamOp[T, []T]

Window groups items into overlapping sliding windows.

func WithContext

func WithContext[T any](ctx context.Context) StreamOp[T, T]

WithContext stops an operation when ctx is canceled before or during iteration. It preserves upstream errors and downstream early stop.