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

chunkflow

package module
v0.0.1 Latest Latest
Warning

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

 Go to latest Published: Sep 15, 2026 License: MIT Imports: 6 Imported by: 0

README

ChunkFlow

ChunkFlow is a rigorous, lazily evaluated data pipeline for Go 1.27+. It wraps the native iter.Seq interface, providing a type-safe Fluent API tailored for stream processing, transformations, and data chunking for I/O operations (ETL).

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.
        NewStream(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

Stream[T] and IoStream[T] have exactly one way of being created: wrapping a native Go iterator.

Function Description
NewStream(iter.Seq[T]) Wraps a native iterator into a synchronous Stream[T].
NewIoStream(ctx, iter.Seq[T]) Wraps a native iterator into a context-aware, error-propagating IoStream[T].
NewIoStream2(ctx, iter.Seq2[T, error]) Wraps a (value, error) iterator into an IoStream[T]. Inverse of IoStream.Seq().
Generators (seq sub-package)

Iterator generators live in github.com/shults/chunkflow/seq. They return plain iter.Seq[T], so they work with NewStream, NewIoStream, 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
Intermediate Operations (Lazy Transformations)

These modify the pipeline logic but execute absolutely no work until a terminal operation is invoked. Every method below exists with the same name and shape on both Stream[T] and IoStream[T].

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.
Take Take(int) Limits the stream to the first N elements.
Skip Skip(int) Bypasses the first N elements.
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.

IoStream[T] additionally offers context-aware, error-returning variants and concurrency:

Method Signature Description
MapAsync MapAsync[R](func(ctx, T) (R, error), ...Option) Like Map, but may fail and run on WithParallel(n) workers (order not preserved for n > 1).
FilterAsync FilterAsync(func(ctx, T) (bool, error), ...Option) Like Filter, with the same error and concurrency semantics as MapAsync.
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 default options (WithParallel, WithLogger, WithDiscardLogger) inherited by all downstream operations.
Error semantics in IoStream

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.

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
Function Signature Description
Flatten Flatten[E](Stream[[]E]) Unwraps a stream of slices into a flat stream of elements. Use with .Through().
IoFlatten IoFlatten[E](IoStream[[]E]) IoStream counterpart of Flatten. Use with .Through().
Terminal Operations (Execution Triggers)

These pull the trigger, forcing the intermediate pipeline to evaluate. On IoStream[T] every terminal operation additionally returns an error: the first error produced upstream (or by the callback) stops consumption and is returned.

Method Stream returns IoStream returns Description Warning
Collect []T ([]T, error) Materializes all processed elements into a slice. OOM risk for infinite streams
Reduce T (T, error) Aggregates elements into a single accumulated value. 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 (int, error) Consumes the stream and returns the number of elements. Requires a finite stream
Any bool (bool, error) Short-circuits and returns true on the first match. Empty stream: false. -
All bool (bool, error) Short-circuits and returns false on the first mismatch. Empty stream: true (vacuous truth). -
First (T, bool) (T, bool, error) Retrieves the first element and short-circuits. -
Last (T, bool) (T, bool, error) Consumes the entire stream to return the final element. Requires a finite stream
Seq iter.Seq[T] iter.Seq2[T, error] Exposes the pipeline as a native iterator. -

IoStream[T] also provides ReduceAsync, AllAsync, AnyAsync and ForEachAsync, whose callbacks receive the context.Context and may return an error.

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). It targets stream processing and chunked I/O work (ETL-style batching) and comes in two flavours that share one fluent API:

  • Stream[T] is synchronous and infallible. Every intermediate operation is a closure over the upstream iterator, nothing is buffered except by Chunk, and nothing runs until a terminal operation pulls.
  • IoStream[T] adds a context.Context, error propagation and optional worker pools. Every Stream method exists on IoStream with the same name and shape; terminal operations additionally return an error, and callbacks that need the context or may fail have an *Async counterpart (MapAsync, FilterAsync, ReduceAsync, AllAsync, AnyAsync, ForEachAsync).

Constructing streams

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

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

rows, err := chunkflow.NewIoStream(ctx, seq.Items(ids...)).
	MapAsync(fetchRow, chunkflow.WithParallel(8)).
	Chunk[[]Row](500).
	ForEachAsync(insertBatch)

NewIoStream2 wraps an iter.Seq2[T, error] and is the inverse of IoStream.Seq, which exposes any IoStream as a native (value, error) iterator.

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)
ioStream.Chunk[[]int](3).Through(chunkflow.IoFlatten)

Errors in IoStream

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

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 IoStream a consumer stopping, or the context being cancelled, also shuts down any worker pool spawned by MapAsync or FilterAsync; the cancellation itself is reported as an error by the terminal operation.

Options

WithParallel(n) sets the worker count for *Async operations (default 1, ordered; with n > 1 the output order is not guaranteed). WithLogger and WithDiscardLogger control diagnostics. Options passed to Opts become defaults for every downstream operation; options passed to a single operation apply to that call only.

Index

Examples

Constants

This section is empty.

Variables

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 IoStream.Seq(); the original error stays in the chain, so errors.Is(err, original) keeps working too.

Functions

This section is empty.

Types

type IoStream 

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

IoStream represents a lazily evaluated, context-aware pipeline whose elements may carry errors. It mirrors the Stream API; every method that accepts a user callback has an *Async counterpart that receives a context and may return an error.

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

func IoFlatten 

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

IoFlatten unwraps a stream of slices into a flat stream of individual elements. It is the IoStream counterpart of Flatten; use it with Through.

func NewIoStream 

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

NewIoStream wraps a native Go iterator into a context-aware IoStream. The context is checked before every element is emitted.

Example 
package main

import (
	"context"
	"fmt"
	"strings"

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

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

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

	fmt.Println(upper, err)
}

Output:
[GO ITER SEQ] <nil>

func NewIoStream2 

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

NewIoStream2 wraps a native (value, error) iterator into an IoStream. It is the inverse of IoStream.Seq.

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.NewIoStream2(ctx, rows).Collect()
	fmt.Println(got, err)
}

Output:
[alice bob] connection lost

func (IoStream[T]) All 

func (s IoStream[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 (IoStream[T]) AllAsync 

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

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

func (IoStream[T]) Any 

func (s IoStream[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 (IoStream[T]) AnyAsync 

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

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

func (IoStream[T]) Chunk 

func (s IoStream[T]) Chunk[R []T](size int) IoStream[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.NewIoStream(ctx, 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>

func (IoStream[T]) CircuitBreaker 

func (s IoStream[T]) CircuitBreaker(maxConsecutiveFailures int) IoStream[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) are never suppressed.

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.NewIoStream(ctx, seq.Range(1, 7)).
		MapAsync(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.NewIoStream(ctx, seq.Numbers(1)). // infinite source
								MapAsync(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 (IoStream[T]) Collect 

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

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

func (IoStream[T]) Count 

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

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

func (IoStream[T]) Exec 

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

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

func (IoStream[T]) Filter 

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

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

func (IoStream[T]) FilterAsync 

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

FilterAsync 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 (IoStream[T]) First 

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

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

func (IoStream[T]) ForEach 

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

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

func (IoStream[T]) ForEachAsync 

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

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

func (IoStream[T]) Last 

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

Last consumes the entire stream and returns the final element.

func (IoStream[T]) Map 

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

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

func (IoStream[T]) MapAsync 

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

MapAsync 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.NewIoStream(ctx, seq.Items("1", "2", "x", "4")).
		MapAsync(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.NewIoStream(ctx, seq.RangeInclusive(1, 100)).
		MapAsync(square, chunkflow.WithParallel(4)).
		Reduce(0, func(item, acc int) int { return acc + item })

	fmt.Println(sum, err)
}

Output:
338350 <nil>

func (IoStream[T]) Opts 

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

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

func (IoStream[T]) Reduce 

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

Reduce aggregates the stream into a single value.

func (IoStream[T]) ReduceAsync 

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

ReduceAsync aggregates the stream into a single value using a context-aware function. Reduction is always sequential; WithParallel is accepted for API symmetry and logged.

func (IoStream[T]) Seq 

func (s IoStream[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.NewIoStream(ctx, seq.Range(0, 6)).MapAsync(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 (IoStream[T]) Skip 

func (s IoStream[T]) Skip(nr int) IoStream[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 (IoStream[T]) Take 

func (s IoStream[T]) Take(nr int) IoStream[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 (IoStream[T]) Through 

func (s IoStream[T]) Through[R any](transform func(IoStream[T]) IoStream[R]) IoStream[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 IoFlatten).

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.IoStream[int]) chunkflow.IoStream[int] {
		return s.Filter(func(i int) bool { return i%2 == 0 })
	}

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

	fmt.Println(got, err)
}

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

type Option 

type Option func(*options)

Option configures an IoStream or a single asynchronous operation.

func WithDiscardLogger 

func WithDiscardLogger() Option

WithDiscardLogger drops all diagnostics. Handy to silence a logger previously set via Opts.

func WithLogger 

func WithLogger(logger *slog.Logger) Option

WithLogger sets the logger used for diagnostics. Diagnostics are discarded by default.

func WithParallel 

func WithParallel(concurrency int) Option

WithParallel sets the number of concurrent workers used by *Async operations.

type Stream 

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

Stream represents a lazily evaluated pipeline operating on elements of type T. Transforming operations do not allocate memory until a terminal operation is called.

func Flatten 

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

Flatten unwraps a stream of slices into a continuous, flat stream of individual elements. It is implemented as a top-level function rather than a method because Go's compiler does not support narrowing receiver type constraints (i.e., enforcing that T must be a slice).

Example 
package main

import (
	"fmt"

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

func main() {
	// Flatten changes the element type, so it is a top-level function.
	// Through keeps the pipeline readable left to right.
	flat := chunkflow.NewStream(seq.Range(1, 8)).
		Chunk[[]int](3).
		Through(chunkflow.Flatten).
		Collect()

	fmt.Println(flat)
}

Output:
[1 2 3 4 5 6 7]

func NewStream 

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

NewStream initializes a stream based on a native Go iterator.

Example 
package main

import (
	"fmt"

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

func main() {
	evens := chunkflow.NewStream(seq.Range(0, 10)).
		Filter(func(i int) bool { return i%2 == 0 }).
		Collect()

	fmt.Println(evens)
}

Output:
[0 2 4 6 8]

func (Stream[T]) All 

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

All is a terminal operation that verifies if all elements satisfy the predicate. Returns false and short-circuits on the first mismatch.

On an empty stream All returns true (vacuous truth): there is no element that could violate the predicate. This mirrors slices.ContainsFunc's complement and keeps All(p) == !Any(!p) for every stream.

Example 
package main

import (
	"fmt"

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

func main() {
	s := chunkflow.NewStream(seq.Items[int]())

	// Vacuous truth: nothing in the stream can violate the predicate.
	fmt.Println(s.All(func(int) bool { return false }))
	fmt.Println(s.Any(func(int) bool { return true }))
}

Output:
true
false

func (Stream[T]) Any 

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

Any is a terminal operation that verifies if at least one element satisfies the predicate. Returns true and short-circuits on the first match.

On an empty stream Any returns false: there is no element that could be a witness.

func (Stream[T]) Chunk 

func (a 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. Panics if size < 1.

Example 
package main

import (
	"fmt"

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

func main() {
	chunks := chunkflow.NewStream(seq.Range(1, 8)).
		Chunk(3).
		Collect()

	fmt.Println(chunks)
}

Output:
[[1 2 3] [4 5 6] [7]]

func (Stream[T]) Collect 

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

Collect is a terminal operation that materializes the stream into a slice allocated on the heap. Unsafe for infinite streams.

func (Stream[T]) Count 

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

Count is a terminal operation that consumes the entire stream and returns the number of elements. Unsafe for infinite streams.

func (Stream[T]) Exec 

func (a Stream[T]) Exec()

Exec is a terminal operation that exhausts the stream to force the execution of side effects.

func (Stream[T]) Filter 

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

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

func (Stream[T]) First 

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

First consumes at most one element. Returns the element and true, or a zero value and false if the stream was empty.

func (Stream[T]) ForEach 

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

ForEach is a terminal operation that executes a side effect for each element.

func (Stream[T]) Last 

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

Last consumes the entire stream and returns the final visited element.

func (Stream[T]) Map 

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

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

Example 
package main

import (
	"fmt"

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

func main() {
	words := chunkflow.NewStream(seq.Items("a", "bb", "ccc")).
		Map(func(s string) int { return len(s) }).
		Collect()

	fmt.Println(words)
}

Output:
[1 2 3]

func (Stream[T]) Reduce 

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

Reduce is a terminal operation that aggregates the stream into a single value.

Example 
package main

import (
	"fmt"

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

func main() {
	sum := chunkflow.NewStream(seq.RangeInclusive(1, 10)).
		Reduce(0, func(item, acc int) int { return acc + item })

	fmt.Println(sum)
}

Output:
55

func (Stream[T]) Seq 

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

Seq returns the underlying native Go iterator for integration with standard packages.

func (Stream[T]) Skip 

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

Skip bypasses the first nr elements and emits the remainder of the stream.

func (Stream[T]) Take 

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

Take consumes at most nr elements from the stream and then short-circuits.

Example (Infinite) 
package main

import (
	"fmt"

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

func main() {
	// Numbers is infinite; Take short-circuits so only 5 elements are ever generated.
	first := chunkflow.NewStream(seq.Numbers(1)).
		Map(func(i int) int { return i * i }).
		Take(5).
		Collect()

	fmt.Println(first)
}

Output:
[1 4 9 16 25]

func (Stream[T]) Through 

func (a 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) due to Go's type system constraints.

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.NewStream or chunkflow.NewIoStream, or used directly in range-over-func loops.
 Package seq provides generators of native Go iterators (iter.Seq) that can be fed into chunkflow.NewStream or chunkflow.NewIoStream, 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.