chunkflow

README

ChunkFlow

ChunkFlow is a stream-processing library for Go built on the language's native iterators. It takes the part of the job that every I/O-bound pipeline reinvents: a context that reaches every step, errors that travel with the data instead of aborting it, worker pools that keep the input order, batching with a size and a time bound, retries and circuit breaking. Pipelines are typed end to end and read left to right, because they are built on Go 1.27 generic methods.

err := chunkflow.New(ctx).Chan(events).                   // any iter.Seq, iter.Seq2, (T, error) iterator or channel
    MapCtx(enrich, chunkflow.WithParallel(8)).            // I/O on 8 workers, results in source order
    Through(policy.CircuitBreaker[Enriched](5)).          // tolerate a flaky dependency, trip on 5 in a row
    ChunkTimeout[[]Enriched](100, 30*time.Millisecond).   // batches of 100, or whatever arrived in 30ms
    ForEachCtx(store)                                     // one write per batch

Goals

• One type. Stream[T] carries a context.Context and lets errors flow as elements. There is no second, simpler stream for in-memory work; pass context.Background() and ignore the error you know cannot happen.
• Errors are data, bugs are not. An error from a source or a callback moves down the pipeline like a value. Operators pass it along, terminals stop at the first one, and it is tolerated only when something says so: a policy, or the callback itself with Suppress. Nothing is dropped or logged behind your back. A panic in a callback becomes ErrPanic, which nothing may tolerate.
• Parallel without losing order. WithParallel(n) runs a step on n workers and emits in source order behind a bounded window, so a parallel step is a drop-in for the sequential one. WithUnordered() trades that for throughput when order does not matter.
• Bounded by construction. Nothing runs until a terminal pulls, and nothing is buffered except what you asked for: a chunk's size, a pool's window. Allocations are constant in the stream's length; see BENCHMARK.md for the numbers.
• Native in, native out. Sources are iter.Seq, iter.Seq2 (as Entry key/value pairs), iter.Seq2[T, error] or channels; Seq() hands the pipeline back as an iterator. Everything that produces or consumes Go iterators already works with it.
• Extensible on one seam. Transform exposes the raw (value, error) sequence and Suppress marks an error as tolerated. The policy and step sub-packages are written on that and nothing else, so an extension in your module has the same tools.

Requires Go 1.27 or newer.

go get -u 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
}

Every operation has a runnable example on pkg.go.dev; the examples below are taken from those tests, so they compile and their output is checked.

Real-time event processing

The shape most pipelines end up with: events are produced all over the application, in bursts, and have to be written somewhere in batches without making the producers wait for the database. Wrap the entry point in a struct with Push and Stop, inject the store, and let the stream behind it do the work.

type Event struct {
    ID      int
    Payload string
}

// DB is the dependency injected into the saver: anything that stores a batch.
type DB interface {
    Insert(ctx context.Context, events []Event) error
}

// BatchEventSaver collects events and writes them to the DB in batches: a full batch goes out
// at once, a partial one after maxWait. Push never waits for the database, only for room in
// the buffer. A failed insert loses that batch and is reported through the error callback; the
// saver itself keeps running. That is the trade-off of this architecture, chosen on purpose.
type BatchEventSaver struct {
    events  chan Event
    done    chan struct{}
    onError atomic.Pointer[func(error)]
}

func NewBatchEventSaver(ctx context.Context, db DB, size int, maxWait time.Duration) *BatchEventSaver {
    s := &BatchEventSaver{events: make(chan Event, 1024), done: make(chan struct{})}
    go s.run(ctx, db, size, maxWait)
    return s
}

func (s *BatchEventSaver) Push(e Event)             { s.events <- e }          // safe for concurrent use
func (s *BatchEventSaver) Stop()                    { close(s.events); <-s.done } // waits for the last batch
func (s *BatchEventSaver) SetOnError(fn func(error)) { s.onError.Store(&fn) }

func (s *BatchEventSaver) run(ctx context.Context, db DB, size int, maxWait time.Duration) {
    defer close(s.done)
    report := func(err error) {
        if fn := s.onError.Load(); fn != nil {
            (*fn)(err)
        }
    }
    insert := func(ctx context.Context, batch []Event) error {
        return chunkflow.Suppress(db.Insert(ctx, batch)) // a lost batch is reported, not fatal
    }
    err := chunkflow.New(ctx).Chan(s.events).
        Opts(chunkflow.WithOnError(report)).          // every tolerated error reaches the callback
        ChunkTimeout[[]Event](size, maxWait).         // batches of size, or whatever arrived in maxWait
        TapCtx(insert).                               // the write; its error is suppressed above
        Drain()                                       // runs until Stop closes the channel
    if err != nil {
        report(err) // only the context can end this pipeline early
    }
}

Three events, a pause longer than maxWait, two more, and a database that refuses the second batch:

saver := NewBatchEventSaver(ctx, db, 100, 30*time.Millisecond)
saver.SetOnError(func(err error) { fmt.Println("lost:", err) })

for _, id := range []int{1, 2, 3} {
    saver.Push(Event{ID: id})
}
time.Sleep(300 * time.Millisecond)
for _, id := range []int{4, 5} {
    saver.Push(Event{ID: id})
}
saver.Stop()
// insert [1 2 3]                            <- released by the 30ms clock, not by the size
// lost: suppressed: insert [4 5]: db down   <- the insert failed; reported, and the saver went on
// stopped

Every line of run is one decision, and each is replaceable: Chunk instead of ChunkTimeout for a source that never pauses; step.Decorate(insert, step.Retry(3)) before giving up on a batch; MapCtx(enrich, WithParallel(8)) before the batching to call a service per event with the order kept; policy.CircuitBreaker to stop hammering a database that is down. The full program is Example_batchEventSaver in example_event_test.go, and its output is checked by go test.

Batching

Chunk(size) groups values into slices and is free of goroutines: it emits when a chunk is full or when the source ends. ChunkTimeout(size, maxWait) also releases a partial chunk once maxWait has passed since its first value, which is what a trickling source needs. Both keep the guarantee that a chunk has at least one value, so batch[0] is always safe. Errors pass through both immediately; the partial chunk waits for the next value.

events := make(chan string)
go func() {
    defer close(events)
    for _, e := range []string{"a", "b", "c"} {
        events <- e
    }
    time.Sleep(300 * time.Millisecond)
    for _, e := range []string{"d", "e"} {
        events <- e
    }
}()

batches, err := chunkflow.New(ctx).Chan(events).ChunkTimeout[[]string](10, 30*time.Millisecond).Collect()
fmt.Println(batches, err) // [[a b c] [d e]] <nil>

Flatten is the inverse (Chunk(3).Through(chunkflow.Flatten)), and Chunk before a MapCtx that takes a slice is how you turn per-item I/O into bulk I/O.

Errors: fail fast, tolerate on purpose

A pipeline without a policy is fail-fast: the first error stops the terminal and the source is not read past the failing element. Three things change that, each explicit.

A callback that knows better returns Suppress(err). The element is skipped, the pipeline continues, and the error is still visible to WithOnError and Seq():

parse := func(_ context.Context, s string) (int, error) {
    n, err := strconv.Atoi(s)
    return n, chunkflow.Suppress(err) // Suppress(nil) is nil; a bad record is not a reason to stop
}

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) // [1 3 5] <nil> 2

A policy on the stream decides for a whole stage. policy.CircuitBreaker(n) tolerates up to n-1 consecutive errors, marking them ErrSuppressed, and trips on the n-th with a fatal error:

got, err := chunkflow.New(ctx).Seq(seq.Range(1, 7)).
    MapCtx(fetch).                                // fails for 3 and 4
    Through(policy.CircuitBreaker[int](3)).       // two in a row are tolerated, a third would trip
    Collect()
fmt.Println(got, err) // [10 20 50 60] <nil>

A decorator on the call wraps the callback you do not own: step.Tolerate(errNotFound) suppresses a specific error, step.Retry repeats a failure. See Retries and timeouts.

Whatever tolerated an error, Seq() shows it, and the switch below is the whole vocabulary:

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.

Parallelism and order

WithParallel(n) on MapCtx, FilterCtx or TapCtx runs the callback on n workers. Results come out in source order, behind a read-ahead window of 2n elements, so the parallel stage is a drop-in for the sequential one:

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

squares, err := chunkflow.New(ctx).Seq(seq.RangeInclusive(1, 6)).
    MapCtx(square, chunkflow.WithParallel(4)).
    Collect()
fmt.Println(squares, err) // [1 4 9 16 25 36] <nil>

Ordering has a price: one slow element holds back its successors, and once the window is full the workers wait. When the consumer does not care, WithUnordered() emits results as they finish, which is the right call for side-effect pipelines ending in Drain:

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) // 338350 <nil>

A pool costs about half a microsecond per element in handoffs, so it pays once the callback costs more than a few microseconds, which every network or disk call does; below that, Map on one goroutine is faster. Eight workers on 200 µs calls give a 7.9× speed-up with order kept, within 1% of unordered. Numbers in BENCHMARK.md.

Retries and timeouts per call

Policies see the stream; the step package sees the individual call. A step.Middleware wraps one invocation and knows only its context and its error, so one set of them serves every callback shape: Decorate for func(ctx, T) (R, error), Decorate3 for fold callbacks. The first middleware in the list is the outermost.

fetch := func(_ context.Context, id int) (string, error) { /* flaky */ }

users, err := chunkflow.New(ctx).Seq(seq.Items(1, 2)).
    MapCtx(step.Decorate(fetch,
        step.Retry(3),                  // up to three attempts...
        step.Timeout(time.Second),      // ...each with its own one-second deadline
    )).
    Collect()

Retry never repeats a call whose caller context is done or whose error was marked by Suppress, and a deadline set by an inner Timeout is not the caller's context, so a timed-out attempt is retried. Pacing is injected, not built in: step.RetryWithBackoff(newBackoff) takes a constructor for anything with the NextBackOff() / Reset() method set of cenkalti/backoff, one fresh instance per element, no dependency taken. Retries and a breaker compose: three attempts per element, then a breaker over five failed elements in a row.

Joining streams

Zip(other, fn) pairs two streams position by position through a callback and ends at the shorter side; a third source is another Zip whose callback extends your struct, so fields keep their names. Concat runs streams one after another, Merge concurrently as they arrive.

ids := chunkflow.New(ctx).Seq(seq.Items(7, 8, 9))
names := chunkflow.New(ctx).Seq(seq.Items("ann", "bob")) // shorter: the pipeline ends with it

type user struct {
    ID   int
    Name string
}
users, err := ids.Zip(names, func(id int, name string) user { return user{ID: id, Name: name} }).Collect()
fmt.Println(users, err) // [{7 ann} {8 bob}] <nil>

Iterators in and out

New(ctx).Seq(...) wraps any iter.Seq[T]; Seq2 any two-value iterator such as maps.All or slices.All, as a stream of Entry{Key, Value}; SeqErr an iter.Seq2[T, error] such as a database cursor, whose errors become the stream's errors; Chan a receive channel. Seq() on a stream returns iter.Seq2[T, error] for a plain range loop, and Transform lets a function rewrite that raw sequence and hand it back as a stream with the same context and options. The seq sub-package has the usual generators (Items, Range, Numbers, Repeat, Iterate) as plain iter.Seq, so they work with slices.Collect and range as well as here.

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[K, V])	Wraps a two-value iterator (maps.All, slices.All) as Stream[Entry[K, V]].
.SeqErr(iter.Seq2[T, error])	Wraps a (value, error) iterator, the inverse of Stream.Seq(); the errors become the stream's errors.
.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.
ChunkTimeout	ChunkTimeout[R []T](size, maxWait)	Chunk that also releases a partial chunk once maxWait has passed since its first value: batching for sources that trickle. Reads the source on a goroutine, so it costs a channel handoff per value.
Zip / ZipCtx	Zip[O, R](other Stream[O], func(T, O) R) · ZipCtx[O, R](other, func(ctx, T, O) (R, error))	Pairs values position by position through the callback, ends at the shorter side. Errors pass at their position without consuming a value on the other side. Context merged like Merge. A third source is another Zip extending your struct.
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.

Step decorators in the step sub-package wrap the callback itself instead of the stream, so they see the individual call and can repeat or bound it, which a policy cannot. A step.Middleware sees only the context and the error of one call, values stay in a closure, so one set of them serves every callback shape without type arguments. Decorate applies them to a func(ctx, T) (R, error), Decorate3 to a fold callback func(ctx, Acc, T) (Acc, error); the first middleware in the list is the outermost.

step.Retry(attempts, opts...)	Repeats a failed call up to attempts times, immediately or paced by step.RetryWithBackoff(newBackoff), where the constructor returns anything with the NextBackOff() / Reset() method set of cenkalti/backoff, created fresh per element. The package ships no pacing of its own and takes no dependency. Never past a done caller context or an error marked by Suppress; a timeout from an inner step.Timeout is retried.
step.Timeout(d)	Runs the call with a context that expires after d. Cooperative: the callback must honour its context.
step.Tolerate(targets...)	Suppress for callbacks you do not own: an error matching any target skips the element. In Decorate that is the marked error; in Decorate3 the fold keeps its accumulator and carries on.

users.MapCtx(step.Decorate(fetch, step.Retry(3, step.RetryWithBackoff(newBackoff)), step.Timeout(2*time.Second)),
        chunkflow.WithParallel(8)).
    Through(policy.CircuitBreaker[User](5))   // 3 attempts of 2s per element, then a breaker over 5 elements in a row

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.

Performance

Measured, not claimed: BENCHMARK.md. The short version: a sequential stage costs about 10 ns per element and a constant handful of allocations per stream, a parallel stage about half a microsecond per element in channel handoffs, so WithParallel pays once the callback costs more than a few microseconds, which every I/O call does. Eight workers on 200 µs calls give a 7.9× speed-up with source order kept, within 1% of WithUnordered().

Versioning

Pre-1.0. A patch release changes no exported API; a minor release adds API or breaks it, and every break is listed in CHANGELOG.md with the migration. v0.1.0 freezes the core API: from there on a break is a deliberate minor bump with a migration note, not a routine.

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
make bench   # every benchmark 8 times into bench.out, summarised with benchstat

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 any two-value iterator such as maps.All as a stream of Entry{Key, Value}, SeqErr an iter.Seq2[T, error] (the inverse of Stream.Seq, its errors become the stream's), 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.

Example (BatchEventSaver)

package main

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	"github.com/shults/chunkflow"
)

// Event is what the application emits.
type Event struct {
	ID      int
	Payload string
}

// DB is the dependency injected into the saver: anything that stores a batch.
type DB interface {
	Insert(ctx context.Context, events []Event) error
}

// BatchEventSaver collects events and writes them to the DB in batches: a full batch goes out
// at once, a partial one after maxWait. Push never waits for the database, only for room in the
// buffer. A failed insert loses that batch and is reported through the error callback; the saver
// itself keeps running. That is the trade-off of this architecture, chosen on purpose here.
type BatchEventSaver struct {
	events  chan Event
	done    chan struct{}
	onError atomic.Pointer[func(error)]
}

func NewBatchEventSaver(ctx context.Context, db DB, size int, maxWait time.Duration) *BatchEventSaver {
	s := &BatchEventSaver{events: make(chan Event, 1024), done: make(chan struct{})}
	go s.run(ctx, db, size, maxWait)
	return s
}

// Push queues one event. Safe for concurrent use.
func (s *BatchEventSaver) Push(e Event) { s.events <- e }

// Stop closes the queue and waits until the last batch has been written.
func (s *BatchEventSaver) Stop() {
	close(s.events)
	<-s.done
}

// SetOnError installs the callback that receives every failed insert (and the reason the saver
// stopped, if it stopped because of the context). May be called at any time.
func (s *BatchEventSaver) SetOnError(fn func(error)) { s.onError.Store(&fn) }

func (s *BatchEventSaver) run(ctx context.Context, db DB, size int, maxWait time.Duration) {
	defer close(s.done)
	report := func(err error) {
		if fn := s.onError.Load(); fn != nil {
			(*fn)(err)
		}
	}
	insert := func(ctx context.Context, batch []Event) error {
		return chunkflow.Suppress(db.Insert(ctx, batch)) // a lost batch is reported, not fatal
	}
	err := chunkflow.New(ctx).Chan(s.events).
		Opts(chunkflow.WithOnError(report)).
		ChunkTimeout[[]Event](size, maxWait).
		TapCtx(insert).
		Drain()
	if err != nil {
		report(err) // only the context can end this pipeline early
	}
}

// fakeDB prints what it stores and refuses one batch.
type fakeDB struct{}

func (fakeDB) Insert(_ context.Context, batch []Event) error {
	ids := make([]int, 0, len(batch))
	for _, e := range batch {
		ids = append(ids, e.ID)
	}
	if ids[0] == 4 {
		return fmt.Errorf("insert %v: db down", ids)
	}
	fmt.Println("insert", ids)
	return nil
}

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

	saver := NewBatchEventSaver(ctx, fakeDB{}, 100, 30*time.Millisecond)
	saver.SetOnError(func(err error) { fmt.Println("lost:", err) })

	// A burst of three events, a pause longer than maxWait, two more, then Stop.
	for _, id := range []int{1, 2, 3} {
		saver.Push(Event{ID: id, Payload: "..."})
	}
	time.Sleep(300 * time.Millisecond)
	for _, id := range []int{4, 5} {
		saver.Push(Event{ID: id, Payload: "..."})
	}
	saver.Stop()
	fmt.Println("stopped")
}

Output:
insert [1 2 3]
lost: suppressed: insert [4 5]: db down
stopped

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[K, V any](seq iter.Seq2[K, V]) Stream[Entry[K, V]]
  • func (b Builder) SeqErr[T any](seq iter.Seq2[T, error]) Stream[T]
• type Entry
• 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]) ChunkTimeout[R []T](size int, maxWait time.Duration) 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]
  • func (s Stream[T]) Zip[O, R any](other Stream[O], fn func(a T, b O) R) 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]

Examples

• Package (BatchEventSaver)
• Builder.Chan
• Builder.Seq2
• Builder.SeqErr
• Merge
• New
• Stream.Chunk
• Stream.Chunk (Deduplication)
• Stream.ChunkTimeout
• Stream.First
• Stream.First (Find)
• Stream.MapCtx
• Stream.MapCtx (Parallel)
• Stream.ReduceBy
• Stream.Seq
• Stream.TapCtx
• Stream.Through
• Stream.Zip
• 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, SeqErr 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[K, V any](seq iter.Seq2[K, V]) Stream[Entry[K, V]]

Seq2 wraps a native two-value iterator, such as maps.All or slices.All, as a stream of Entry values: the first value becomes Key, the second Value. For an iterator whose second value is an error, use SeqErr, which turns it into the stream's own error channel instead. The context is checked before every element, as in Seq.

Example

package main

import (
	"context"
	"fmt"
	"maps"
	"slices"

	"github.com/shults/chunkflow"
)

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

	// A map streams as Entry values; maps.All is a plain iter.Seq2[K, V].
	stock := map[string]int{"apples": 3, "pears": 0, "plums": 7}
	inStock, err := chunkflow.New(ctx).Seq2(maps.All(stock)).
		Filter(func(e chunkflow.Entry[string, int]) bool { return e.Value > 0 }).
		Map(func(e chunkflow.Entry[string, int]) string { return e.Key }).
		Collect()
	slices.Sort(inStock) // map order is random
	fmt.Println(inStock, err)
}

Output:
[apples plums] <nil>

func (Builder) SeqErr

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

SeqErr wraps a native (value, error) iterator, the inverse of Stream.Seq: the error of each pair becomes the element's error, so a failing source ends the pipeline exactly like a failing callback. 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).SeqErr(rows).Collect()
	fmt.Println(got, err)
}

Output:
[alice bob] connection lost

type Entry

type Entry[K, V any] struct {
	Key   K
	Value V
}

Entry is one element of a stream built with Seq2: a key/value pair as yielded by maps.All, slices.All (index and element) or any other iter.Seq2[K, V]. It is plain data with no behaviour, the one exported pair in the package: a map entry has exactly two named sides and never nests, unlike the pairs a Zip would build, which is why Zip takes a callback instead.

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]) ChunkTimeout

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

ChunkTimeout is Chunk with a bound on how long a value may wait for its chunk: a chunk is emitted when it reaches size or when maxWait has passed since its first value arrived, whichever comes first. The clock starts with the first value of a chunk, so an idle source never produces empty chunks. It is the batching operator for sources that trickle, such as a channel fed by producers, where Chunk would hold a partial batch until the source speaks again.

Unlike Chunk it needs a goroutine: a pull-based iterator only gets control when the source yields, so the source is read on a feeder goroutine and the emitter selects between the next value and the timer. That costs a channel handoff per value, and, as with every operator that reads the source from a goroutine, a consumer that stops early cannot interrupt a source that is blocked inside its own pull; a Chan source blocked on its channel is released when the channel or its context gives up, not by Take. Errors pass through immediately, the partial chunk is kept and its clock keeps running. The stream's cancellation surfaces as an error even when the feeder exits without emitting. size below 1 or a non-positive maxWait emits a single error and ends.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/shults/chunkflow"
)

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

	// Events trickle in over a channel: three quickly, then a pause, then two more.
	events := make(chan string)
	go func() {
		defer close(events)
		for _, e := range []string{"a", "b", "c"} {
			events <- e
		}
		time.Sleep(300 * time.Millisecond)
		for _, e := range []string{"d", "e"} {
			events <- e
		}
	}()

	// Batches of up to 10, but no event waits longer than 30ms for its batch: the first
	// three go out together long before the pause is over.
	batches, err := chunkflow.New(ctx).Chan(events).ChunkTimeout[[]string](10, 30*time.Millisecond).Collect()
	fmt.Println(batches, err)
}

Output:
[[a b c] [d e]] <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.

func (Stream[T]) Zip

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

Zip pairs the values of this stream with the values of other, position by position, and emits fn(a, b) for each pair. It ends when either side ends; a value already pulled from the longer side is dropped, as in the iter.Zip proposal for the standard library. Errors are not paired: an error on either side is forwarded at its position without consuming a value from the other side, so `Zip` obeys the same rule as every operator, values are what it works on.

The result keeps this stream's options and a context merged like Merge does: values and deadline from this stream, cancellation from either. other is driven through iter.Pull, so it runs one coroutine that stops with the pipeline; other may itself be any pipeline, parallel stages included. Pairing is a pure function; for I/O on the pair use ZipCtx or a following MapCtx. A third source is another Zip whose fn extends the struct built by the first:

ids.Zip(users, func(id ID, u User) Row { return Row{ID: id, User: u} }).
    Zip(orders, func(r Row, o Order) Row { r.Order = o; return r })

Example

package main

import (
	"context"
	"fmt"

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

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

	ids := chunkflow.New(ctx).Seq(seq.Items(7, 8, 9))
	names := chunkflow.New(ctx).Seq(seq.Items("ann", "bob")) // shorter: the pipeline ends with it

	type user struct {
		ID   int
		Name string
	}
	users, err := ids.Zip(names, func(id int, name string) user { return user{ID: id, Name: name} }).Collect()
	fmt.Println(users, err)
}

Output:
[{7 ann} {8 bob}] <nil>

func (Stream[T]) ZipCtx

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

ZipCtx is Zip with a context-aware fn that may fail; a returned error takes the pair's position in the stream and, unless suppressed, ends it at the terminal. The context fn receives is the merged one described on Zip.