chunkflow

README

ChunkFlow

Lazy, type-safe data pipelines on Go's native iterators, with what I/O-bound stream processing needs on top: a context, errors that travel as elements, worker pools, a circuit breaker and chunking. Built on Go 1.27 generic methods, so the whole pipeline reads left to right.

err := chunkflow.New(ctx).Chan(userIDs).                 // any iter.Seq, iter.Seq2 or channel
    MapCtx(fetchUser, chunkflow.WithParallel(8)).        // I/O on 8 workers
    Through(policy.CircuitBreaker[User](5)).             // tolerate flaky lookups, trip on 5 in a row
    Filter(func(u User) bool { return u.Active }).
    Chunk[[]User](500).                                  // batch for the database
    ForEachCtx(insertBatch)                              // one INSERT per 500 users

Why

• One type. Stream[T] carries a context.Context and an error channel. Pure in-memory work passes context.Background() and ignores the error it knows cannot happen.
• Lazy and bounded. Every intermediate operation is a closure over the upstream iterator; nothing runs until a terminal pulls, and nothing is buffered except by Chunk (bounded by its size) and by worker pools (bounded by twice WithParallel). Take, First, Any and All stop the source as soon as they can.
• Parallel, still in order. WithParallel(n) runs a step on n workers and emits results in source order, so MapCtx(f, WithParallel(8)).Collect() returns what the sequential version would, only faster. One slow item holds back its successors; WithUnordered() gives that up for throughput when the consumer does not care about order.
• Errors are data. An error from a source or a callback flows down the pipeline as an element. Operators pass it along and keep working on values; terminals stop at the first one. An error is tolerated only when something says so, a policy such as policy.CircuitBreaker or a callback returning Suppress(err), and it is marked rather than dropped, so nothing is lost.
• Extensible where it matters. Transform hands the raw (value, error) sequence to any function and Suppress marks an error as tolerated; that is the whole seam. The policy package is written on it and nothing else, so a policy in your module has the same tools.
• Bugs are not data. A panic in a callback, also on a worker goroutine, becomes an ErrPanic error that nothing may suppress. Whether a step runs on one worker or eight makes no difference to how a bug surfaces.
• Native iterators in and out. Sources are iter.Seq, iter.Seq2[T, error] or channels; Seq() gives the pipeline back as iter.Seq2[T, error]. Generators in seq are plain iter.Seq, usable with slices.Collect and range too.

Requires Go 1.27+ (generic methods, iter).

go get github.com/shults/chunkflow

Quick start

package main

import (
    "context"
    "errors"
    "fmt"

    "github.com/shults/chunkflow"
    "github.com/shults/chunkflow/policy"
    "github.com/shults/chunkflow/seq"
)

func main() {
    ctx := context.Background()

    // Squares of the first even numbers, in batches of 3, flattened back. Numbers is infinite;
    // Take decides how much of it is ever generated.
    res, err := chunkflow.New(ctx).Seq(seq.Numbers(1)).
        Filter(func(i int) bool { return i%2 == 0 }).
        Map(func(i int) int { return i * i }).
        Take(7).
        Chunk[[]int](3).
        Through(chunkflow.Flatten). // Flatten changes the element type, so it is a function
        Collect()
    fmt.Println(res, err) // [4 16 36 64 100 144 196] <nil>

    // A fallible step. The first error ends the pipeline; values collected so far are returned.
    parse := func(_ context.Context, s string) (int, error) {
        var n int
        _, err := fmt.Sscanf(s, "%d", &n)
        return n, err
    }
    nums, err := chunkflow.New(ctx).Seq(seq.Items("1", "2", "x", "4")).MapCtx(parse).Collect()
    fmt.Println(nums, err != nil) // [1 2] true

    // Tolerating errors: policy.CircuitBreaker(3) lets two failures in a row through and trips on
    // the third. Policies enter through Through; the element type cannot be inferred from the
    // threshold and is spelled out. Tolerated errors stay visible through Seq() or WithOnError.
    tolerated := 0
    nums, err = chunkflow.New(ctx).Seq(seq.Items("1", "x", "3", "y", "5")).
        Opts(chunkflow.WithOnError(func(err error) {
            if errors.Is(err, chunkflow.ErrSuppressed) {
                tolerated++
            }
        })).
        MapCtx(parse).
        Through(policy.CircuitBreaker[int](3)).
        Collect()
    fmt.Println(nums, err, tolerated) // [1 3 5] <nil> 2
}

More runnable examples, one per operation, are in *_example_test.go and on pkg.go.dev.

API

Building a stream

New binds the context; the source method picks the element type. Options come afterwards, through Opts on the stream or on the individual *Ctx call.

New(ctx)	Starts a Stream bound to ctx.
.Seq(iter.Seq[T])	Wraps a native iterator. The context is checked before every element.
.Seq2(iter.Seq2[T, error])	Wraps a (value, error) iterator, the inverse of Stream.Seq().
.Chan(<-chan T)	Reads a channel until it is closed or the context is cancelled. Single-use; stopping early does not close or drain the channel.

Generators in github.com/shults/chunkflow/seq return plain iter.Seq[T]:

seq.Items(...T)	the arguments	finite
seq.Range(from, to)	integers in [from, to), any integer type	finite
seq.RangeInclusive(from, to)	integers in [from, to], safe at MaxInt	finite
seq.Repeat(val, n)	val exactly n times	finite
seq.Numbers(start)	start, start+1, ...	infinite
seq.Const(val)	val forever	infinite
seq.Iterate(seed, fn)	seed, fn(seed), fn(fn(seed)), ...	infinite

Anything else that yields iter.Seq plugs in directly: slices.Values, maps.Keys, strings.Lines, a database cursor wrapped as iter.Seq2[Row, error].

Intermediate operations

Lazy; nothing runs until a terminal pulls. Each *Ctx variant takes a callback that receives the context.Context and may return an error. The plain variant is the *Ctx one on a single worker.

Method	Signature	Notes
Map / MapCtx	Map[R](func(T) R) · MapCtx[R](func(ctx, T) (R, error), ...StepOption)	WithParallel(n) runs the callback on n workers, results in source order; WithUnordered() emits them as they arrive.
Filter / FilterCtx	Filter(func(T) bool) · FilterCtx(func(ctx, T) (bool, error), ...StepOption)	Same concurrency semantics as MapCtx.
Tap / TapCtx	Tap(func(T)) · TapCtx(func(ctx, T) error, ...StepOption)	Side effect, element passes through; an error from TapCtx replaces the element.
Take / Skip	Take(n) · Skip(n)	Count values, not errors. Take stops pulling from the source.
TakeWhile / SkipWhile	TakeWhile(func(T) bool) · SkipWhile(func(T) bool), *Ctx variants	Stop / start emitting at the first false; SkipWhile stops evaluating afterwards.
CompactFunc	CompactFunc(func(a, b T) bool)	Drops consecutive duplicates in O(1) memory; input must be sorted or grouped for a global dedup.
Chunk	Chunk[R []T](size)	Groups values into slices of size; the last one may be shorter. Errors pass through, the partial chunk is kept.
Through	Through[R](func(Stream[T]) Stream[R])	Plugs a top-level function into the chain, keeping left-to-right order.
Opts	Opts(...Option)	Pipeline options for everything downstream.

Top-level functions

Operations that change the element type or need a constraint a method cannot express. Use them with .Through().

Flatten[E](Stream[[]E]) Stream[E]	Unwraps a stream of slices.
Compact[T comparable](Stream[T])	CompactFunc with ==.
Concat(...Stream[T])	One stream after another, deterministic; errors keep their position.
Merge(...Stream[T])	All streams concurrently, interleaved as they arrive; cancelled by any source's context, with the original cause kept.
Suppress(err) error	Marks an error as tolerated, inside a *Ctx callback or a policy (see below).

Policies and extensions

Error policies are not operations on values, so they are functions for .Through() in the policy sub-package. The element type cannot be inferred from a policy's arguments and is spelled out.

policy.CircuitBreaker[T](n)	Tolerates up to n-1 errors in a row by re-emitting them marked ErrSuppressed; trips on the n-th. Never tolerates context errors or ErrPanic.

Writing your own: Stream.Transform[R](func(iter.Seq2[T, error]) iter.Seq2[R, error]) Stream[R] gives a function the raw element sequence, fatal errors included and without stopping at them, and wraps the result back into a stream with the same context and options. Mark what you tolerate with Suppress. policy.CircuitBreaker uses nothing else.

Terminal operations

Every terminal returns an error: the first non-suppressed error stops consumption and is returned, together with whatever was produced so far.

Method	Returns	Notes
Collect()	([]T, error)	Everything into a slice. Do not use on infinite streams.
ForEach / ForEachCtx	error	Side effect per element.
Reduce[R](init R, func(acc R, item T) R) / ReduceCtx	(R, error)	Fold into an accumulator of any type, (acc, item) order, init first. Always sequential.
ReduceBy[K, R](key func(T) K, init R, func(acc R, item T) R) / ReduceByCtx	(map[K]R, error)	One fold per key: group (nil, append), count (0, +1), sum, max. init is copied per key, so start from a scalar or nil. Materialises like Collect.
Count()	(int, error)
Drain()	error	Exhausts the stream, discards values; the terminal for side-effect pipelines.
All / AllCtx	(bool, error)	Short-circuits on the first mismatch. Empty stream: true (vacuous truth).
Any / AnyCtx	(bool, error)	Short-circuits on the first match. Empty stream: false.
First()	(T, error)	Stops the source after one value. Empty stream: ErrEmpty.
Last()	(T, error)	Empty stream: ErrEmpty.
Seq()	iter.Seq2[T, error]	The pipeline as a native iterator, suppressed errors included; stops after the first fatal one.

Options

Two kinds, checked by the compiler:

• StepOption configures one *Ctx call: WithParallel(n) and WithUnordered(). Passing it to Opts makes it the default for everything downstream.
  • WithParallel(n) keeps source order. The step reads ahead of the slowest item by a window of 2n items; when the window is full, idle workers wait instead of pulling, so memory stays bounded and a single slow item stalls the step behind it (head-of-line blocking).
  • WithUnordered() drops the ordering and the window: results leave as workers finish them, and a downstream policy.CircuitBreaker counts consecutive errors in arrival order.
• Option configures the whole pipeline only: WithOnError(fn) registers the hook every terminal calls for each error it handles, suppressed ones just before skipping them, the fatal one just before returning it. It is the place for logging and metrics.

Invalid values (WithParallel(0), WithOnError(nil)) do not panic: the stream they are applied to emits a single error and ends.

Errors, in one place

for v, err := range stream.Through(policy.CircuitBreaker[User](5)).Seq() {
    switch {
    case err == nil:
        use(v)
    case errors.Is(err, chunkflow.ErrSuppressed):
        metrics.Inc("tolerated")  // the original error is still in the chain
    case errors.Is(err, chunkflow.ErrPanic):
        log.Fatal(err)            // a bug in a callback; message has the value and the stack
    default:
        return err                // breaker tripped, context cancelled, source failed, ...
    }
}

• Operators work on values: Take(3) takes three values however many errors pass by, Chunk never puts an error into a slice, Compact compares only values.
• A pipeline without a policy is fail-fast: the source is not consumed past the failing element.
• A callback that knows an error is tolerable returns Suppress(err): the element is replaced by the marked error, terminals skip it, WithOnError and Seq() still see it, policy.CircuitBreaker does not count it. Suppress(nil) is nil; context errors and ErrPanic cannot be suppressed and come back unchanged.
• Cancelling the context always surfaces as an error, also when a worker pool exits without emitting.
• First and Last on a stream without values return ErrEmpty. It is a plain sentinel for "nothing to return", not a failure: it never reaches WithOnError, and errors.Is(err, ErrEmpty) tells it apart from the errors above.

Development

make setup   # installs golangci-lint and govulncheck into ./bin, enables the git pre-commit hook
make check   # gofmt, go vet, golangci-lint, go mod tidy -diff — what the hook runs
make test    # go test -race -shuffle=on
make ci      # check + test + govulncheck, mirrors the GitHub Actions pipeline

Design decisions and their reasons are in CONVENTIONS.md, the plan in ROADMAP.md, contributor instructions in AGENTS.md.

Documentation

Overview

Package chunkflow provides lazily evaluated, type-safe data pipelines on top of Go's native iterators (iter.Seq), aimed at stream processing and chunked I/O work (ETL-style batching). One type, Stream[T], carries a context.Context, propagates errors as elements and can run steps on worker pools. Every intermediate operation is a closure over the upstream iterator, nothing is buffered except by Chunk, and nothing runs until a terminal operation pulls. Callbacks that need the context or may fail have a *Ctx variant (MapCtx, FilterCtx, ReduceCtx, ReduceByCtx, AllCtx, AnyCtx, ForEachCtx); the plain variant is that *Ctx variant pinned to one worker.

Constructing streams

Streams are built from native iterators only. Generators live in the sub-package seq (Items, Range, RangeInclusive, Numbers, Const, Repeat, Iterate) so that they are equally usable with slices.Collect, range loops and this package:

evens, err := chunkflow.New(ctx).Seq(seq.Range(0, 100)).
	Filter(func(i int) bool { return i%2 == 0 }).
	Collect()

rows, err := chunkflow.New(ctx).Seq(seq.Items(ids...)).
	MapCtx(fetchRow, chunkflow.WithParallel(8)).
	Chunk[[]Row](500).
	ForEachCtx(insertBatch)

New(ctx) binds the context first and lets the source pick the element type: Seq wraps an iter.Seq, Seq2 an iter.Seq2[T, error] (the inverse of Stream.Seq), Chan a receive channel.

Type-changing operations

Go methods cannot introduce type parameters that constrain the receiver, so operations such as Flatten (Stream[[]E] -> Stream[E]) are top-level functions. Through keeps the left-to-right reading order when using them:

stream.Chunk[[]int](3).Through(chunkflow.Flatten)

Errors

An error, whether produced by the source or by a callback, travels down the pipeline as an element. Intermediate operations never interpret it: Map, Filter, Take, Skip, Chunk and Flatten forward it untouched and keep working on the values around it, so Take(n) and Skip(n) count values only. Terminal operations stop at the first error they see, which makes a plain pipeline fail fast and stop consuming the source right after the failing element.

Errors are tolerated only when something says so. A callback returns Suppress(err), replacing that one element by an error matching ErrSuppressed with the original in the chain; terminal operations skip such elements, consumers of Seq can still observe them. Policies do the same for whole streams: policy.CircuitBreaker[T](n), plugged in with Through, tolerates errors below its threshold and trips with a fatal error on the n-th consecutive one. Context errors are never suppressed.

Extensions

Transform hands the raw (value, error) sequence, fatal errors included, to any function and wraps the result back into a Stream with the same context and options; Suppress is the only way to mark an error as tolerated. The policy package is written on exactly these two and nothing else, so an extension in another module has the same tools.

A panic inside a callback is recovered where it happens, also on worker goroutines, and becomes an error matching ErrPanic that nothing may suppress. A bug therefore surfaces the same way whether the step runs sequentially or on WithParallel(n) workers.

Short-circuiting and cancellation

Every operation honours the yield protocol: when a consumer stops (Take, First, Any, All, or a plain break in a range loop), the whole chain stops pulling from the source. In Stream a consumer stopping, or the context being cancelled, also shuts down any worker pool spawned by MapCtx or FilterCtx; the cancellation itself is reported as an error by the terminal operation.

Options

Two kinds exist. A StepOption configures one *Ctx call: WithParallel(n) sets the worker count of MapCtx, FilterCtx or TapCtx (default 1; results keep source order, at the cost of a bounded reorder window), WithUnordered lets such a step emit results as they arrive instead. An Option configures the whole pipeline through Opts and is inherited downstream: every StepOption also works as an Option (a default worker count), and WithOnError(fn) registers the hook that terminal operations call for every error they handle, suppressed ones included, so logging and metrics need no manual loop over Seq.

Index

• Variables
• func Suppress(err error) error
• type Builder
  • func New(ctx context.Context) Builder
  • func (b Builder) Chan[T any](ch <-chan T) Stream[T]
  • func (b Builder) Seq[T any](seq iter.Seq[T]) Stream[T]
  • func (b Builder) Seq2[T any](seq iter.Seq2[T, error]) Stream[T]
• type Option
  • func WithOnError(fn func(error)) Option
• type StepOption
  • func WithParallel(concurrency int) StepOption
  • func WithUnordered() StepOption
• type Stream
  • func Compact[T comparable](stream Stream[T]) Stream[T]
  • func Concat[T any](streams ...Stream[T]) Stream[T]
  • func Flatten[E any](stream Stream[[]E]) Stream[E]
  • func Merge[T any](streams ...Stream[T]) Stream[T]
  • func (s Stream[T]) All(predicate func(T) bool) (bool, error)
  • func (s Stream[T]) AllCtx(predicate func(ctx context.Context, item T) (bool, error)) (bool, error)
  • func (s Stream[T]) Any(predicate func(T) bool) (bool, error)
  • func (s Stream[T]) AnyCtx(predicate func(ctx context.Context, item T) (bool, error)) (bool, error)
  • func (s Stream[T]) Chunk[R []T](size int) Stream[R]
  • func (s Stream[T]) Collect() ([]T, error)
  • func (s Stream[T]) CompactFunc(eq func(a, b T) bool) Stream[T]
  • func (s Stream[T]) Count() (int, error)
  • func (s Stream[T]) Drain() error
  • func (s Stream[T]) Filter(predicate func(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]) First() (T, error)
  • func (s Stream[T]) ForEach(fn func(T)) error
  • func (s Stream[T]) ForEachCtx(fn func(ctx context.Context, item T) error) error
  • func (s Stream[T]) Last() (T, error)
  • func (s Stream[T]) Map[R any](mapFn func(T) R) Stream[R]
  • 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]) Reduce[R any](init R, fn func(acc R, item T) R) (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]) 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]) ReduceCtx[R any](init R, fn func(ctx context.Context, acc R, item T) (R, error)) (R, error)
  • func (s Stream[T]) Seq() iter.Seq2[T, error]
  • func (s Stream[T]) Skip(nr int) Stream[T]
  • 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]) Take(nr int) 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]
  • func (s Stream[T]) Through[R any](transform func(Stream[T]) Stream[R]) Stream[R]
  • func (s Stream[T]) Transform[R any](transform func(iter.Seq2[T, error]) iter.Seq2[R, error]) Stream[R]

Examples

• Builder.Chan
• Builder.Seq2
• Merge
• New
• Stream.Chunk
• Stream.Chunk (Deduplication)
• Stream.First
• Stream.First (Find)
• Stream.MapCtx
• Stream.MapCtx (Parallel)
• Stream.ReduceBy
• Stream.Seq
• Stream.TapCtx
• Stream.Through
• Suppress
• WithUnordered

Variables

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

ErrEmpty is returned by First and Last when the stream produced no value. It is a plain sentinel, not a wrapped element error: nothing failed, there was nothing to return. It is never passed to the WithOnError hook.

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

ErrPanic marks an error produced from a panic inside a user callback of an Stream (MapCtx, FilterCtx, ForEachCtx, ...). The panic is recovered where it happens, also on worker goroutines, and travels down the pipeline as an error element that nothing may suppress: Suppress refuses it, policies pass it through and every terminal returns it. The message carries the panic value and the stack of the goroutine that panicked; if the value is an error it is also in the chain, so errors.Is(err, original) keeps working.

var ErrSuppressed = errors.New("suppressed")

ErrSuppressed marks an error that something decided to tolerate: a policy such as policy.CircuitBreaker, or a callback that returned Suppress(err). Terminal operations skip elements carrying such an error instead of failing. Use errors.Is(err, ErrSuppressed) to detect one, for example when iterating Stream.Seq(); the original error stays in the chain, so errors.Is(err, original) keeps working too.

Functions

func Suppress

func Suppress(err error) error

Suppress marks err as tolerated. Return it from a MapCtx, FilterCtx or TapCtx callback when the failure of this one element is not a reason to stop the pipeline: the element is replaced by the marked error, terminal operations skip it, WithOnError and Seq still see it with err in the chain, and a downstream policy.CircuitBreaker does not count it. Policies use the same function: it is the only way to produce a tolerated error.

Suppress(nil) is nil, so `return v, chunkflow.Suppress(err)` needs no branch. An error that is already suppressed is returned as is. Context errors and ErrPanic cannot be suppressed; they come back unchanged and stay fatal, which a policy detects with errors.Is(Suppress(err), ErrSuppressed).

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"strconv"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// A malformed record is not a reason to stop the import; the parser says so itself.
	parse := func(_ context.Context, s string) (int, error) {
		n, err := strconv.Atoi(s)
		return n, chunkflow.Suppress(err) // Suppress(nil) is nil
	}

	skipped := 0
	nums, err := chunkflow.New(ctx).Seq(seq.Items("1", "x", "3", "", "5")).
		Opts(chunkflow.WithOnError(func(err error) {
			if errors.Is(err, chunkflow.ErrSuppressed) {
				skipped++
			}
		})).
		MapCtx(parse).
		Collect()

	fmt.Println(nums, err, skipped)
}

Output:
[1 3 5] <nil> 2

Types

type Builder

type Builder struct {
	// contains filtered or unexported fields
}

Builder binds a context before the element type is known. Obtain one with New and turn it into a Stream with Seq, Seq2 or Chan; each of those is a generic method, so the element type is inferred from the source.

func New

func New(ctx context.Context) Builder

New starts building a Stream bound to ctx. Pipeline options are set on the stream with Opts, step options on the individual *Ctx call:

chunkflow.New(ctx).Chan(jobs).Opts(chunkflow.WithParallel(8)).MapCtx(process).Drain()

Example

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	upper, err := chunkflow.New(ctx).Seq(seq.Items("go", "iter", "seq")).
		Map(strings.ToUpper).
		Collect()

	fmt.Println(upper, err)
}

Output:
[GO ITER SEQ] <nil>

func (Builder) Chan

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

Chan reads from ch until it is closed or the context is cancelled; a cancellation ends the stream with the context error as its last element.

The resulting stream is single-use: a channel cannot be rewound, so iterating the stream twice yields whatever the first pass left behind. Stopping early (Take, First, a break) stops reading but does not close ch or signal the producer; a producer that may outlive the consumer must watch the same context.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// A producer that respects the same context: if the consumer stops early the
	// producer is released too, instead of blocking forever on a send.
	jobs := make(chan int)
	go func() {
		defer close(jobs)
		for i := 1; ; i++ {
			select {
			case jobs <- i:
			case <-ctx.Done():
				return
			}
		}
	}()

	firstThree, err := chunkflow.New(ctx).Chan(jobs).
		Map(func(i int) int { return i * i }).
		Take(3).
		Collect()

	fmt.Println(firstThree, err)
}

Output:
[1 4 9] <nil>

func (Builder) Seq

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

Seq wraps a native Go iterator. The context is checked before every element is emitted; once it is cancelled the next element carries the context error.

func (Builder) Seq2

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

Seq2 wraps a native (value, error) iterator, the inverse of Stream.Seq. Elements whose error is nil still pick up the context error once the context is cancelled.

Example

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/shults/chunkflow"
)

func main() {
	ctx := context.Background()

	// A (value, error) iterator, e.g. wrapping a database cursor.
	rows := func(yield func(string, error) bool) {
		for _, r := range []string{"alice", "bob"} {
			if !yield(r, nil) {
				return
			}
		}
		yield("", errors.New("connection lost"))
	}

	got, err := chunkflow.New(ctx).Seq2(rows).Collect()
	fmt.Println(got, err)
}

Output:
[alice bob] connection lost

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option configures a whole pipeline. Pass it to Opts; every operation downstream inherits it. Every StepOption is also an Option.

func WithOnError

func WithOnError(fn func(error)) Option

WithOnError registers a callback that terminal operations invoke for every error they handle: errors marked ErrSuppressed just before skipping them, and the fatal error just before returning it. It is the hook for logging or metrics; it cannot alter the outcome. A nil callback is invalid: the stream it is applied to emits a single error and ends.

type StepOption

type StepOption func(*options)

StepOption configures a single *Ctx call, e.g. MapCtx(fn, WithParallel(8)). Passing it to Opts instead makes it the default for everything downstream.

func WithParallel

func WithParallel(concurrency int) StepOption

WithParallel sets the number of concurrent workers used by MapCtx, FilterCtx and TapCtx. Results are emitted in source order regardless of n, so a parallel step is indistinguishable from a sequential one except for speed; see WithUnordered to trade that for throughput. A value below 1 is invalid: the stream it is applied to emits a single error and ends.

Ordering has a price. A finished item waits for its slower predecessors, and the step reads ahead of the slowest one by a bounded window (currently 2n items pulled from the source and not yet emitted); once the window is full, idle workers wait rather than pull. One very slow item therefore stalls the whole step behind it, with memory bounded by the window.

func WithUnordered

func WithUnordered() StepOption

WithUnordered lets a parallel step (WithParallel(n > 1)) emit results as workers finish them instead of in source order. Use it when the consumer does not care about order, typically side-effect pipelines ending in Drain, to avoid head-of-line blocking: no result ever waits for a slower predecessor. A downstream policy.CircuitBreaker then counts consecutive errors in arrival order. It has no effect on a single worker.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// When only the total matters, WithUnordered lets fast items overtake slow ones instead
	// of waiting for them; the fold below is order-independent, so nothing is lost.
	square := func(_ context.Context, i int) (int, error) { return i * i, nil }
	sum, err := chunkflow.New(ctx).Seq(seq.RangeInclusive(1, 100)).
		MapCtx(square, chunkflow.WithParallel(4), chunkflow.WithUnordered()).
		Reduce(0, func(acc, item int) int { return acc + item })

	fmt.Println(sum, err)
}

Output:
338350 <nil>

type Stream

type Stream[T any] struct {
	// contains filtered or unexported fields
}

Stream is a lazily evaluated, context-aware pipeline over a native iterator whose elements may carry errors. Nothing runs until a terminal operation pulls. Every method that accepts a user callback has an *Ctx variant whose callback receives a context and may return an error; the plain variant is that *Ctx variant pinned to one worker.

Error semantics: an error produced by the source or by a callback travels down the pipeline as an element. Intermediate operations pass it through untouched and keep processing the remaining input; they operate on values only and never interpret errors. A policy such as policy.CircuitBreaker, or a callback returning Suppress(err), may mark an error as tolerated (see ErrSuppressed). Terminal operations skip elements whose error matches ErrSuppressed and stop at the first other error, so a pipeline that tolerates nothing behaves as "fail fast".

A panic inside any user callback is recovered where it happens, also on worker goroutines, and becomes an error matching ErrPanic that no operator may suppress. Whether a step runs sequentially or on WithParallel(n) workers therefore makes no difference to how a bug surfaces: the terminal returns it.

func Compact

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

Compact drops consecutive duplicate values using ==. It requires comparable T, which a method cannot demand, so it is a top-level function; use it with Through. See Stream.CompactFunc for the semantics and the sorted-or-grouped input precondition.

func Concat

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

Concat emits every element of the first stream, then of the second, and so on. Order is deterministic and a stream is not touched until all previous ones are exhausted. Errors pass through at their position. The result inherits the options of the first stream and a context that is cancelled when any source context is (see mergeContexts); with no arguments it returns an empty stream.

func Flatten

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

Flatten unwraps a stream of slices into a flat stream of individual elements. It changes the element type from []E to E, which a method cannot express, so it is a top-level function; use it with Through.

func Merge

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

Merge consumes all streams concurrently, one goroutine per source, and emits elements as they arrive. The interleaving is not deterministic. Errors pass through at their arrival position. When the consumer stops, or when the context of any source is cancelled, every source goroutine is released and the cancellation is reported as an error element carrying the original cause. The result inherits the options of the first stream and the merged context; with no arguments it returns an empty stream.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// Two independent sources (think: two shards) consumed concurrently. The interleaving is
	// not deterministic, so aggregate instead of printing the order.
	shardA := chunkflow.New(ctx).Seq(seq.Range(0, 50))
	shardB := chunkflow.New(ctx).Seq(seq.Range(50, 100))

	n, err := chunkflow.Merge(shardA, shardB).Count()
	fmt.Println(n, err)
}

Output:
100 <nil>

func (Stream[T]) All

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

All verifies whether all elements satisfy the predicate. Short-circuits on the first mismatch.

On an empty stream, or one consisting only of suppressed errors, All returns true (vacuous truth): no element could violate the predicate. All(p) == !Any(!p) always holds.

func (Stream[T]) AllCtx

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

AllCtx verifies whether all elements satisfy the context-aware predicate.

func (Stream[T]) Any

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

Any verifies whether at least one element satisfies the predicate. Short-circuits on the first match.

On an empty stream, or one consisting only of suppressed errors, Any returns false: no element could be a witness.

func (Stream[T]) AnyCtx

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

AnyCtx verifies whether at least one element satisfies the context-aware predicate.

func (Stream[T]) Chunk

func (s Stream[T]) Chunk[R []T](size int) Stream[R]

Chunk groups elements into slices of the given size. The final chunk may contain fewer elements than size. Emits an error if size < 1. Errors are passed through immediately; the partially filled chunk is kept and continues to accumulate subsequent elements.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// Chunk turns N+1 round trips into N/size batches.
	err := chunkflow.New(ctx).Seq(seq.Range(1, 8)).
		Chunk(3).
		ForEach(func(batch []int) {
			fmt.Println("insert", batch)
		})

	fmt.Println(err)
}

Output:
insert [1 2 3]
insert [4 5 6]
insert [7]
<nil>

Example (Deduplication)

ExampleStream_Chunk_deduplication shows the recommended way to drop duplicates on a large stream: batch with Chunk, ask the store once per batch, flatten back. The library deliberately has no Distinct operator, because a global "seen" set has to live somewhere with O(unique) memory, and that somewhere should be the user's choice (a database, a key-value store, a Bloom filter, ...), not a hidden map inside the pipeline.

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// Stand-in for an external store, e.g. an `INSERT ... ON CONFLICT DO NOTHING RETURNING id`.
	store := map[string]struct{}{}
	insertNew := func(_ context.Context, batch []string) ([]string, error) {
		var fresh []string
		for _, id := range batch {
			if _, dup := store[id]; dup {
				continue
			}
			store[id] = struct{}{}
			fresh = append(fresh, id)
		}
		return fresh, nil // one round trip per batch, not per element
	}

	err := chunkflow.New(ctx).Seq(seq.Items("a", "b", "a", "c", "b", "d", "a")).
		Chunk[[]string](3).
		MapCtx(insertNew).
		Through(chunkflow.Flatten).
		ForEach(func(id string) { fmt.Println("new:", id) })

	fmt.Println(err)
}

Output:
new: a
new: b
new: c
new: d
<nil>

func (Stream[T]) Collect

func (s Stream[T]) Collect() ([]T, error)

Collect materializes the stream into a slice. Elements collected before an error are returned together with that error.

func (Stream[T]) CompactFunc

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

CompactFunc drops consecutive duplicate values: a value is emitted only if eq reports it different from the previously emitted value. Like slices.CompactFunc it removes duplicates only when they are adjacent, so the input must be sorted or grouped for a global de-duplication. Memory is O(1). Errors pass through and do not reset the comparison: a value, an error, then the same value yields the value once.

func (Stream[T]) Count

func (s Stream[T]) Count() (int, error)

Count consumes the entire stream and returns the number of elements seen before the first error.

func (Stream[T]) Drain

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

Drain exhausts the stream, discarding values, and returns the first error. It is the terminal for pipelines whose work happens in TapCtx or MapCtx side effects.

func (Stream[T]) Filter

func (s Stream[T]) Filter(predicate func(T) bool) Stream[T]

Filter emits only the elements for which the predicate returns true.

func (Stream[T]) FilterCtx

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

FilterCtx emits only the elements for which the context-aware predicate returns true. With WithParallel(n > 1) predicates are evaluated by n workers, in source order unless WithUnordered is given, and after a fatal error the pool may still evaluate the predicate on buffered items before it shuts down (see MapCtx). The first error terminates the stream.

func (Stream[T]) First

func (s Stream[T]) First() (T, error)

First returns the first value and stops pulling from the source. It returns ErrEmpty when the stream ends without a value, and the fatal error if one comes first.

Example

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// First stops the source after one value; Numbers is infinite.
	v, err := chunkflow.New(ctx).Seq(seq.Numbers(1)).Filter(func(i int) bool { return i%5 == 0 }).First()
	fmt.Println(v, err)

	// An empty stream is not a failure, it is ErrEmpty.
	_, err = chunkflow.New(ctx).Seq(seq.Items[int]()).First()
	fmt.Println(errors.Is(err, chunkflow.ErrEmpty))
}

Output:
5 <nil>
true

Example (Find)

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// "Find the first match" is Filter followed by First; the source stops right there.
	pulled := 0
	v, err := chunkflow.New(ctx).Seq(seq.Numbers(1)).
		Tap(func(int) { pulled++ }).
		Filter(func(i int) bool { return i%7 == 0 }).
		First()

	fmt.Println(v, err, pulled)
}

Output:
7 <nil> 7

func (Stream[T]) ForEach

func (s Stream[T]) ForEach(fn func(T)) error

ForEach executes a side effect for each element and returns the first error.

func (Stream[T]) ForEachCtx

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

ForEachCtx executes a context-aware side effect for each element and returns the first error.

func (Stream[T]) Last

func (s Stream[T]) Last() (T, error)

Last consumes the entire stream and returns its final value. It returns ErrEmpty when the stream ends without a value, and the fatal error if one occurs.

func (Stream[T]) Map

func (s Stream[T]) Map[R any](mapFn func(T) R) Stream[R]

Map transforms each element of the stream using the provided function.

func (Stream[T]) MapCtx

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

MapCtx transforms each element using a context-aware function that may fail. With WithParallel(n > 1) elements are processed by n workers; results, errors included, still come out in source order unless WithUnordered is given (see WithParallel for the cost of ordering). The first error terminates the stream.

Shutdown is not instantaneous: when a terminal stops because of a fatal error (or a consumer breaks out of Seq), workers that are already inside mapFn finish that call, and items pulled from the source into the pool's buffer may still be handed to mapFn before the pool notices the cancellation. Up to n calls may therefore run after the terminal has returned; their results are discarded. A callback with side effects must tolerate this, or check ctx before acting.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	parse := func(_ context.Context, s string) (int, error) {
		var n int
		_, err := fmt.Sscanf(s, "%d", &n)
		return n, err
	}

	// The first error stops the pipeline; values collected before it are returned.
	nums, err := chunkflow.New(ctx).Seq(seq.Items("1", "2", "x", "4")).
		MapCtx(parse).
		Collect()

	fmt.Println(nums, err != nil)
}

Output:
[1 2] true

Example (Parallel)

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	square := func(_ context.Context, i int) (int, error) { return i * i, nil }

	// Four workers, results still in source order: the parallel step is a drop-in for the
	// sequential one.
	squares, err := chunkflow.New(ctx).Seq(seq.RangeInclusive(1, 6)).
		MapCtx(square, chunkflow.WithParallel(4)).
		Collect()

	fmt.Println(squares, err)
}

Output:
[1 4 9 16 25 36] <nil>

func (Stream[T]) Opts

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

Opts returns a copy of the stream with the given options applied as defaults for all subsequent operations.

func (Stream[T]) Reduce

func (s Stream[T]) Reduce[R any](init R, fn func(acc R, item T) R) (R, error)

Reduce folds the stream into a single value: starting from init, fn is called as fn(acc, item) for every value and its result becomes the next accumulator. The accumulator type R is independent of the element type. On error the accumulator built so far is returned together with the error.

func (Stream[T]) ReduceBy

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)

ReduceBy folds the stream into one accumulator per key: key picks the group of each value, and fn(acc, item) runs the fold of that group exactly as Reduce would, starting from init. It is the map-reduce terminal: grouping (init nil, fn appends), counting (init 0, fn adds one), sums, maxima and the like are all one call. Groups are kept in source order within each key. Values are materialised like Collect, so the result is bounded by the number of keys and the size of their accumulators, not by the stream.

init is copied into every group by value. Start from a scalar, nil or a zero value and let fn allocate; a non-nil map, slice or pointer given as init would be shared between the groups. On error the map built so far is returned together with the error.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	words := seq.Items("go", "is", "fun", "and", "go", "is", "fast")

	// Grouping is a fold that appends, starting from nil.
	groups, err := chunkflow.New(ctx).Seq(words).
		ReduceBy(func(w string) int { return len(w) }, nil, func(acc []string, w string) []string {
			return append(acc, w)
		})
	fmt.Println(groups, err)

	// Counting is a fold that adds one, starting from zero.
	counts, err := chunkflow.New(ctx).Seq(words).
		ReduceBy(func(w string) string { return w }, 0, func(n int, _ string) int { return n + 1 })
	fmt.Println(counts, err)
}

Output:
map[2:[go is go is] 3:[fun and] 4:[fast]] <nil>
map[and:1 fast:1 fun:1 go:2 is:2] <nil>

func (Stream[T]) ReduceByCtx

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)

ReduceByCtx is ReduceBy with a context-aware fn that may fail; a returned error stops the fold and is returned with the map built so far. key stays a plain function: a key is a property of the value, work that needs the context belongs in a preceding MapCtx. Like ReduceCtx it is sequential and takes no StepOption.

func (Stream[T]) ReduceCtx

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

ReduceCtx is Reduce with a context-aware fn that may fail; a returned error stops the fold and is returned with the accumulator built so far. A fold is sequential by nature, so unlike MapCtx it takes no StepOption.

func (Stream[T]) Seq

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

Seq returns the stream as a native (value, error) iterator. Tolerated elements are yielded with an error matching ErrSuppressed; iteration stops after the first error that is not suppressed. To keep going past one, see Transform.

Example

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/policy"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()
	errOdd := errors.New("odd")

	rejectOdd := func(_ context.Context, i int) (int, error) {
		if i%2 == 1 {
			return 0, errOdd
		}
		return i, nil
	}

	// Seq exposes suppressed errors instead of hiding them, so the consumer can
	// count or log what the breaker tolerated.
	tolerated := 0
	for v, err := range chunkflow.New(ctx).Seq(seq.Range(0, 6)).MapCtx(rejectOdd).Through(policy.CircuitBreaker[int](2)).Seq() {
		switch {
		case err == nil:
			fmt.Println("value", v)
		case errors.Is(err, chunkflow.ErrSuppressed):
			tolerated++
		default:
			fmt.Println("fatal", err)
		}
	}
	fmt.Println("tolerated", tolerated)
}

Output:
value 0
value 2
value 4
tolerated 3

func (Stream[T]) Skip

func (s Stream[T]) Skip(nr int) Stream[T]

Skip bypasses the first nr elements and emits the remainder of the stream. Errors are passed through and do not count towards nr.

func (Stream[T]) SkipWhile

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

SkipWhile drops values as long as the predicate returns true and then emits every remaining element without evaluating the predicate again. Errors are passed through and are not evaluated by the predicate.

func (Stream[T]) SkipWhileCtx

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

SkipWhileCtx is SkipWhile with a context-aware predicate. A predicate error is emitted as an error element; the stream continues and the predicate keeps being evaluated on subsequent values.

func (Stream[T]) Take

func (s Stream[T]) Take(nr int) Stream[T]

Take consumes at most nr elements from the stream and then short-circuits. Errors are passed through and do not count towards nr.

func (Stream[T]) TakeWhile

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

TakeWhile emits values as long as the predicate returns true and short-circuits at the first value for which it returns false; nothing further is pulled from the source. Errors are passed through and are not evaluated by the predicate.

func (Stream[T]) TakeWhileCtx

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

TakeWhileCtx is TakeWhile with a context-aware predicate. A predicate error is emitted as an error element; the stream continues and the predicate keeps being evaluated on subsequent values, so a downstream policy can tolerate it.

func (Stream[T]) Tap

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

Tap invokes fn for every value and passes it through unchanged. Errors flow past untouched; fn never sees them. Meant for side effects such as logging or metrics.

func (Stream[T]) TapCtx

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

TapCtx invokes a context-aware fn for every value and passes the value through unchanged. An error returned by fn replaces the value with that error in the stream. With WithParallel(n > 1) fn runs on n workers with the ordering and shutdown behaviour documented on MapCtx; the latter matters most here, because TapCtx exists for side effects. WithUnordered is the natural companion when the order of those side effects is irrelevant.

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// TapCtx can veto an element by returning an error, e.g. an audit check.
	audit := func(_ context.Context, amount int) error {
		if amount > 100 {
			return fmt.Errorf("amount %d exceeds limit", amount)
		}
		return nil
	}

	got, err := chunkflow.New(ctx).Seq(seq.Items(10, 50, 500, 20)).
		TapCtx(audit).
		Collect()

	fmt.Println(got, err)
}

Output:
[10 50] amount 500 exceeds limit

func (Stream[T]) Through

func (s Stream[T]) Through[R any](transform func(Stream[T]) Stream[R]) Stream[R]

Through pipes the current stream into an external transformation function. It acts as a structural bridge to maintain fluent API chaining for operations that must be implemented as top-level functions (like Flatten) and for error policies (like policy.CircuitBreaker).

Example

package main

import (
	"context"
	"fmt"

	"github.com/shults/chunkflow"
	"github.com/shults/chunkflow/seq"
)

func main() {
	ctx := context.Background()

	// Reusable pipeline fragments are plain functions; Through plugs them in.
	onlyEven := func(s chunkflow.Stream[int]) chunkflow.Stream[int] {
		return s.Filter(func(i int) bool { return i%2 == 0 })
	}

	got, err := chunkflow.New(ctx).Seq(seq.Range(0, 10)).
		Through(onlyEven).
		Chunk[[]int](2).
		Through(chunkflow.Flatten).
		Collect()

	fmt.Println(got, err)
}

Output:
[0 2 4 6 8] <nil>

func (Stream[T]) Transform

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

Transform is the extension seam: it hands the raw element sequence to transform as a native iter.Seq2[T, error] and wraps what comes back into a Stream that keeps this stream's context and options. Unlike Seq, the raw sequence does not stop after a fatal error; every element, values and errors alike, reaches transform, which may forward, replace, drop or reorder them and may end the sequence early by returning. Together with Suppress this is all an error policy needs; policy.CircuitBreaker is written on it and nothing else.

transform runs once, when the stream is built, and must only assemble the returned iterator; state that has to start fresh on every iteration belongs inside that iterator. Code in the returned iterator is not a callback: a panic there is not turned into ErrPanic, it propagates. A nil transform yields a single error and ends.