Help us shape the future of Go! Take the 2026 Go Developer Survey

chunkflow

package module
v0.0.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 16, 2026 License: MIT Imports: 7 Imported by: 0

README

ChunkFlow

ChunkFlow is a lazily evaluated data pipeline for Go 1.27+. It wraps native iter.Seq iterators in a type-safe fluent API built on generic methods, and adds what I/O-bound stream processing (ETL) needs on top: a context, errors that travel as elements, worker pools, a circuit breaker and chunking.

Architecture & Constraints

  • O(1) Memory Complexity: All intermediate transformations (except Chunk) are pure closures. They do not allocate heap memory and operate on the stream on the fly.
  • Short-circuiting: Terminal operations like Take, Any, All, or First instantly halt the consumption of the source stream, preventing wasted CPU cycles.
  • Zero Reflection: The package relies entirely on Go 1.27+ generics and generic methods (with necessary compiler workarounds detailed below).
  • Requirements: Go 1.27+ or higher (due to the strict dependency on the native iter package).

Installation

go get github.com/shults/chunkflow

Quick Start

A practical example demonstrating lazy evaluation, chunking (e.g., to prevent N+1 database query problems), and flattening the stream back to its original shape:

package main

import (
    "fmt"
    "github.com/shults/chunkflow"
    "github.com/shults/chunkflow/seq"
)

func main() {

    res := chunkflow.
        New(seq.Numbers(1)). // 1. Wrap an infinite iterator; Skip(5) and Take(20) limit consumption to exactly 25 elements.
        Skip(5).
        Take(20).
        Filter(func(i int) bool { // 2. Lazy filtering on the fly.
            return i%2 == 0
        }).
        Chunk[[]int](3). // 3. Buffer elements into chunks of 3.
        Through(chunkflow.Flatten[int]). // 4. Unwrap the chunks back into a flat stream using the architectural bridge.
        Collect() // 5. Materialize the final result into a physical slice.

    fmt.Println(res)
}

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 — what the hook runs (~0.5s warm)
make test    # go test -race -shuffle=on
make ci      # check + test + govulncheck, mirrors the GitHub Actions pipeline

The pre-commit hook lives in .githooks/ and is enabled via core.hooksPath. Skip it once with git commit --no-verify.

API Reference

Constructors

A Stream[T] is created through a small builder: New binds the context and default options first, and the source method picks the element type.

Function Description
New(ctx, ...Option) Starts a Stream bound to ctx; options become the pipeline defaults.
New(ctx).Seq(iter.Seq[T]) Wraps a native iterator. The context is checked before every element.
New(ctx).Seq2(iter.Seq2[T, error]) Wraps a (value, error) iterator. Inverse of Stream.Seq().
New(ctx).Chan(<-chan T) Reads a channel until it is closed or the context is cancelled. Single-use; stopping early does not close the channel.
Generators (seq sub-package)

Iterator generators live in github.com/shults/chunkflow/seq. They return plain iter.Seq[T], so they work with New(ctx).Seq, slices.Collect and range loops alike.

Function Description Length
seq.Items(...T) Yields the variadic arguments. Finite
seq.Range(from, to) Yields integers in the half-open interval [from, to). Works for any integer type. Finite
seq.RangeInclusive(from, to) Yields integers in the closed interval [from, to]. Safe for to == MaxInt. Finite
seq.Numbers(start) Monotonically increasing integers starting at start. Infinite
seq.Const(val) Repeatedly emits the same value. Infinite
seq.Repeat(val, n) Emits the same value n times; the finite form of Const. Finite
seq.Iterate(seed, fn) Emits seed, fn(seed), fn(fn(seed)), ... Restarts from seed on every pass. Infinite
Intermediate Operations (Lazy Transformations)

These modify the pipeline logic but execute absolutely no work until a terminal operation is invoked.

Method Signature Description
Map Map[R](func(T) R) Transforms each element from type T to R.
Filter Filter(func(T) bool) Emits only elements that satisfy the predicate.
Tap Tap(func(T)) Runs a side effect for each element and passes it through unchanged.
Take Take(int) Limits the stream to the first N elements.
Skip Skip(int) Bypasses the first N elements.
TakeWhile TakeWhile(func(T) bool) Emits elements while the predicate holds, then stops pulling from the source.
SkipWhile SkipWhile(func(T) bool) Drops elements while the predicate holds, then emits the rest without testing.
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](int) Groups elements into physical slices of the given size.
Through Through[R](func(Stream) Stream) Pipes the stream through an external top-level function.

Context-aware, error-returning variants and concurrency:

Method Signature Description
MapCtx MapCtx[R](func(ctx, T) (R, error), ...StepOption) Like Map, but may fail and run on WithParallel(n) workers (order not preserved for n > 1).
FilterCtx FilterCtx(func(ctx, T) (bool, error), ...StepOption) Like Filter, with the same error and concurrency semantics as MapCtx.
TapCtx TapCtx(func(ctx, T) error, ...StepOption) Like Tap; a returned error replaces the element. Same concurrency semantics as MapCtx.
TakeWhileCtx / SkipWhileCtx (func(ctx, T) (bool, error)) Context-aware predicates; always sequential. Errors in the stream pass through unevaluated.
CircuitBreaker CircuitBreaker(maxConsecutiveFailures int) Tolerates up to n-1 errors in a row by re-emitting them marked as ErrSuppressed; trips on the n-th. Context errors are never suppressed.
Opts Opts(...Option) Sets pipeline options inherited by all downstream operations: WithParallel(n) as the default worker count, WithOnError(fn) as the hook terminals call for every error they handle (suppressed ones before skipping, the fatal one before returning).

WithParallel is a StepOption and may also be passed to a single MapCtx / FilterCtx / TapCtx call; WithOnError is pipeline-only, so passing it to a step does not compile. ReduceCtx takes no options: a fold is sequential.

Error semantics

An error from the source or from a callback travels down the pipeline as an element. Intermediate operations pass it through and keep processing the remaining input (Take/Skip do not count errors, Chunk keeps its partial buffer). Terminal operations stop at the first error they see, so a pipeline without a CircuitBreaker fails fast, and the source is not consumed past the failing element.

A panic inside any callback is recovered where it happens, also on worker goroutines, and becomes an error matching ErrPanic (the panic value and the goroutine's stack are in the message; an error value stays in the chain). Nothing may suppress it: CircuitBreaker passes it through and ends, every terminal returns it. Whether a step runs sequentially or on WithParallel(n) workers makes no difference to how a bug surfaces.

CircuitBreaker does not drop errors. Below its threshold it re-emits each error wrapped so that it matches both ErrSuppressed and the original error. Terminal operations skip such elements instead of stopping, so nothing is logged and nothing is lost: iterate Seq() and check errors.Is(err, chunkflow.ErrSuppressed) to observe what was tolerated.

for v, err := range stream.CircuitBreaker(5).Seq() {
    switch {
    case err == nil:
        use(v)
    case errors.Is(err, chunkflow.ErrSuppressed):
        metrics.Inc("tolerated") // original error is still in the chain: errors.Is(err, io.EOF) works
    default:
        return err // fatal: breaker tripped, context cancelled, ...
    }
}
Top-Level Functions

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

Function Signature Description
Flatten Flatten[E](Stream[[]E]) Unwraps a stream of slices into a flat stream of elements.
Compact Compact[T comparable](Stream[T]) CompactFunc with ==. Needs comparable, hence top-level.
Concat Concat(...Stream[T]) Emits the streams one after another, deterministic order; errors keep their position. Name follows slices.Concat.
Merge Merge(...Stream[T]) Consumes all streams concurrently and interleaves elements as they arrive.
Terminal Operations (Execution Triggers)

These pull the trigger, forcing the intermediate pipeline to evaluate. Every terminal operation returns an error: the first error produced upstream (or by the callback) stops consumption and is returned.

Method Returns Description Warning
Collect ([]T, error) Materializes all processed elements into a slice. OOM risk for infinite streams
Reduce[R](init R, fn(acc R, item T) R) (R, error) Folds elements into an accumulator of any type; fn(acc, item) order. Requires a finite stream
ForEach error Executes a side effect for every element. Blocks until completion
Exec error Exhausts the stream, discarding values. -
Count (int, error) Consumes the stream and returns the number of elements. Requires a finite stream
Any (bool, error) Short-circuits and returns true on the first match. Empty stream: false. -
All (bool, error) Short-circuits and returns false on the first mismatch. Empty stream: true (vacuous truth). -
First (T, bool, error) Retrieves the first element and short-circuits. -
Last (T, bool, error) Consumes the entire stream to return the final element. Requires a finite stream
Seq iter.Seq2[T, error] Exposes the pipeline as a native iterator. -

ReduceCtx, AllCtx, AnyCtx and ForEachCtx take callbacks that receive the context.Context and may return an error. Reduce and ReduceCtx are generic in the accumulator type R, so a stream of strings can fold into an int or a map.

Design Notes: Why Flatten requires Through

Due to strict constraints in the Go compiler, it is impossible to apply narrower type constraints to method receivers (e.g., enforcing that T must be a slice []E for a specific method).

To prevent runtime panics caused by empty interface assertions, Flatten is implemented strictly as a top-level function. The Through method acts as a structural bridge to inject this top-level function directly into the pipeline without breaking the Fluent API chain.

// Instead of breaking the chain:
stream2 := chunkflow.Flatten(stream1.Chunk(100))

// Maintain strict left-to-right flow readability:
stream1.Chunk[[]int](100).Through(chunkflow.Flatten[int])

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, 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, chunkflow.WithParallel(8)).Seq(seq.Items(ids...)).
	MapCtx(fetchRow).
	Chunk[[]Row](500).
	ForEachCtx(insertBatch)

New(ctx, opts...) binds the context and default options 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.

CircuitBreaker(n) is the one operator that tolerates errors. Below its threshold it re-emits each error wrapped so that errors.Is(err, ErrSuppressed) is true while the original error stays in the chain; terminal operations skip such elements, and consumers of Seq can still observe them. On the n-th consecutive error the breaker trips with a fatal error. Context errors are never suppressed.

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; with n > 1 the output order of that step is not guaranteed). An Option configures the whole pipeline through New or 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

Examples

Constants

This section is empty.

Variables

View Source 
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: CircuitBreaker passes 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.

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

ErrSuppressed marks an error that a CircuitBreaker decided to tolerate. 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

This section is empty.

Types

type Builder added in v0.0.2 

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

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

func New added in v0.0.2 

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

New starts building an Stream bound to ctx. The options become the defaults of every operation in the resulting pipeline (see Opts): 

chunkflow.New(ctx, chunkflow.WithParallel(8)).Chan(jobs).MapCtx(process).Exec()
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 added in v0.0.2 

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 added in v0.0.2 

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 added in v0.0.2 

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 New or Opts; every operation downstream inherits it. Every StepOption is also an Option.

func WithOnError added in v0.0.2 

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 added in v0.0.2 

type StepOption func(*options)

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

func WithParallel 

func WithParallel(concurrency int) StepOption

WithParallel sets the number of concurrent workers used by MapCtx, FilterCtx and TapCtx. With n > 1 the output order of that step is not guaranteed. A value below 1 is invalid: the stream it is applied to emits a single error and ends.

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. CircuitBreaker 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 without a CircuitBreaker 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 added in v0.0.2 

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 added in v0.0.2 

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 added in v0.0.2 

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 added in v0.0.2 

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 added in v0.0.2 

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]) CircuitBreaker added in v0.0.2 

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

CircuitBreaker tolerates up to maxConsecutiveFailures-1 errors in a row and trips on the next one, interrupting the stream with a wrapped error. A successful element resets the counter.

Tolerated errors are not dropped: they are re-emitted wrapped in an error matching ErrSuppressed so that downstream terminal operations skip them while consumers of Seq() can still observe them. Errors already marked as suppressed by an upstream breaker pass through without affecting the counter. Context errors (context.Canceled, context.DeadlineExceeded) and recovered panics (ErrPanic) are never suppressed: they pass through and end the stream.

After a parallel stage the notion of "consecutive" follows arrival order, not source order.

Example 
package main

import (
	"context"
	"errors"
	"fmt"

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

func main() {
	ctx := context.Background()
	errFlaky := errors.New("flaky")

	fetch := func(_ context.Context, i int) (int, error) {
		if i == 3 || i == 4 {
			return 0, errFlaky
		}
		return i * 10, nil
	}

	// Up to 2 consecutive failures are tolerated; the 3rd in a row would trip the breaker.
	got, err := chunkflow.New(ctx).Seq(seq.Range(1, 7)).
		MapCtx(fetch).
		CircuitBreaker(3).
		Collect()

	fmt.Println(got, err)
}

Output:
[10 20 50 60] <nil>
Example (Tripped) 
package main

import (
	"context"
	"errors"
	"fmt"

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

func main() {
	ctx := context.Background()
	errDown := errors.New("service down")

	fetch := func(_ context.Context, i int) (int, error) {
		if i >= 3 {
			return 0, errDown
		}
		return i, nil
	}

	got, err := chunkflow.New(ctx).Seq(seq.Numbers(1)). // infinite source
								MapCtx(fetch).
								CircuitBreaker(3).
								Collect()

	fmt.Println(got)
	fmt.Println(err)
	fmt.Println(errors.Is(err, errDown))
}

Output:
[1 2]
circuit breaker tripped after 3 consecutive errors: service down
true

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 added in v0.0.2 

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

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

Exec exhausts the stream, discarding values, and returns the first error.

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 added in v0.0.2 

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 and the output order is not guaranteed. The first error terminates the stream.

func (Stream[T]) First 

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

First consumes at most one element. Returns (zero, false, nil) for an empty stream.

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 added in v0.0.2 

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, bool, error)

Last consumes the entire stream and returns the final element.

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 added in v0.0.2 

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 and the output order is not guaranteed. The first error terminates the stream.

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 }

	// With WithParallel(n) the output order is not guaranteed, so reduce instead of collecting.
	sum, err := chunkflow.New(ctx).Seq(seq.RangeInclusive(1, 100)).
		MapCtx(square, chunkflow.WithParallel(4)).
		Reduce(0, func(acc, item int) int { return acc + item })

	fmt.Println(sum, err)
}

Output:
338350 <nil>

func (Stream[T]) Opts added in v0.0.2 

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]) ReduceCtx added in v0.0.2 

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. Elements tolerated by a CircuitBreaker are yielded with an error matching ErrSuppressed; iteration stops after the first error that is not suppressed.

Example 
package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/shults/chunkflow"
	"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).CircuitBreaker(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 added in v0.0.2 

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 added in v0.0.2 

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 added in v0.0.2 

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 added in v0.0.2 

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 CircuitBreaker can tolerate it.

func (Stream[T]) Tap added in v0.0.2 

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 added in v0.0.2 

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 and the output order is not guaranteed, exactly as for MapCtx.

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).

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>

Source Files

View all Source files

Directories

Path Synopsis
Package seq provides generators of native Go iterators (iter.Seq) that can be fed into chunkflow.New(ctx).Seq, or used directly in range-over-func loops.
 Package seq provides generators of native Go iterators (iter.Seq) that can be fed into chunkflow.New(ctx).Seq, or used directly in range-over-func loops.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.