go-stream

English | 简体中文

A Go 1.27 generics implementation of the Java Stream API — built on the newly introduced generic methods feature, enabling natural, fluent stream processing in native Go for the first time:

stream.Of(1, 2, 3, 4, 5).
    Filter(func(n int) bool { return n%2 == 1 }).
    Map(func(n int) int { return n * n }).
    ToSlice() // [1 9 25]

Before Go 1.27, methods could not declare their own type parameters, so chained APIs like Map[U] could only be written as package-level functions (stream.Map(s, f)) with no fluent chaining. With generic methods, methods on Stream[T] can carry their own type parameters, enabling fully chained declarations with static type migration along the pipeline.

Features

• Lazy pipelines: intermediate operations only declare the pipeline without triggering traversal; a terminal operation triggers a single fused evaluation pass
• Generic methods: method-level type parameters such as Map[U]/Zip[U, R]/Collect[A, R] let element types migrate statically along the pipeline
• Single-pass fusion: stateless operators fuse into a single pass (Sink chain) at evaluation time; stateful operators materialize in segments
• Short-circuit evaluation: Limit/First/AnyMatch/TakeWhile and friends stop source traversal as soon as the condition is met (safe for infinite streams)
• Errors as values: expected errors (IO source failures, MapErr family callback errors) propagate as error values — first error short-circuits, partial results are preserved, query via Err(); unrecoverable misuses (double consumption, nil callbacks) panic
• Composition over inheritance: Java's abstract class hierarchy (AbstractPipeline/StatelessOp/StatefulOp) is translated into "struct embedding + constructors + injected function values" with no simulated inheritance
• Java 25 parity highlights: outbound iterator adaptation ToSeq() iter.Seq[T] (range-over-func interop), collector composition ecosystem (GroupingByDownstream/PartitioningBy/Teeing/Filtering/FlatMapping/CollectingAndThen/MinBy/MaxBy), sliding window WindowSliding, single-pass statistics Summary/Summarizing (SummaryStats), and convenience sources RangeClosed/OfNonZero
• Time-window resampling: TimeWindow buckets elements by ts(v).Truncate(d) into aligned tumbling windows (fixed time span instead of fixed element count — Julia/Scala time-window style); per-bucket aggregation is composed via .Map(...) (inline, or FromSlice(b.Items) into any collector)
• Zero third-party dependencies: no third-party runtime dependencies in v1

Installation

go get github.com/JayceChant/go-stream

Requires go 1.27+ (relies on the generic methods feature).

Quick Start

import (
    "github.com/JayceChant/go-stream"
    "github.com/JayceChant/go-stream/collector" // collector subpackage (as needed)
)

// 1. Build from containers (lazy, no traversal yet)
s := stream.FromSlice(data)          // zero-copy reference
r := stream.Range(0, 100)            // integer range [0, 100) → *NumberStream (numeric narrowing, see NumberStream below)
g := stream.Generate(func() int { return 42 }) // infinite generator

// 2. Intermediate operations (return a new Stream, chainable)
s.Filter(p).Map(f).Sorted(cmp).Limit(10)

// 3. Terminal operations (trigger a single evaluation, consume the stream)
s.ToSlice()
s.Count()
s.AnyMatch(p)
s.Collect(collector.GroupingBy(keyOf, valOf))

More runnable examples: example_test.go (verified by go test) and the example/ directory — nine standalone, copy-paste-ready programs covering the full API surface:

go -C example run ./basics      # sources → intermediate → terminal operations
go -C example run ./collectors  # collector family + custom Collector (TopN)
go -C example run ./numeric     # numeric aggregation, Scan, infinite sources, Zip/Chunk/Enumerate
go -C example run ./errors      # errors-as-value model (FromFunc/MapErr family/Err())
go -C example run ./parallel    # Parallel(n)/Unordered, order-preserving merge, auto fallback
go -C example run ./lifecycle   # OnClose/Close resource management, Cache replayable factory
go -C example run ./extensions  # Java 25 parity: ToSeq/collector composition/WindowSliding/Summary/RangeClosed/OfNonZero
go -C example run ./timewindow  # time-window resampling: TimeWindow buckets + Map-composed aggregation
go -C example run ./join        # conditional two-stream joins: Join (inner), LeftJoin, right join via LeftJoin

example/ is a separate Go module (not part of the library's tests or coverage) so each file can be copied into your project as-is.

Implementation Comparison

Style

The same task — keep the positive amounts, sort them in descending order, take the top 3, and format them as price strings. Sorted and Limit are stateful operators that force a materialization point mid-pipeline, and the order is significant: sorting must precede Limit (the top 3 of a sorted sequence), formatting must follow it. Hand-rolled code has no choice but to split into two loops:

Plain Go, no library:

// Plain Go: fine for a one-off loop, but the stateful steps dissolve the
// pipeline into two loops plus an in-place sort.
var amounts []int
for _, n := range orders { // loop 1: only the stateless Filter fits here
    if n > 0 {
        amounts = append(amounts, n)
    }
}
slices.SortFunc(amounts, func(a, b int) int { return b - a }) // materialization point: needs every element (unstable, same contract as Sorted)
var top []string
for i, n := range amounts { // loop 2: top 3 and formatting can only wait here
    if i >= 3 {
        break
    }
    top = append(top, fmt.Sprintf("$%d", n))
}

Functional stream, pre-Go 1.27 (no generic methods):

// Pre-1.27: type-safe and composable, but calls nest and read inside-out —
// the data source ends up buried at the center, reading order opposed to
// execution order.
result := stream.ToSlice( // 5. executed last, written outermost
    stream.Map( // 4. format the top 3
        stream.Limit( // 3. top 3 of the sorted result
            stream.Sorted( // 2. descending sort, forces materialization
                stream.Filter(stream.FromSlice(orders), // 1. the source, read first
                    func(n int) bool { return n > 0 }),
                func(a, b int) int { return b - a },
            ),
            3,
        ),
        func(n int) string { return fmt.Sprintf("$%d", n) },
    ),
)

This library (generic methods):

// Generic methods: reads top-down in pipeline order, element types migrate
// along the chain (int → string), stateful steps slot in seamlessly.
result := stream.FromSlice(orders).
    Filter(func(n int) bool { return n > 0 }).
    Sorted(func(a, b int) int { return b - a }). // stateful: materialize, then sort
    Limit(3). // stateful: top 3 of the sorted result
    Map(func(n int) string { return fmt.Sprintf("$%d", n) }).
    ToSlice()

Style	Pros	Cons
Plain Go	Zero overhead, zero dependencies	The stateful steps force two loops plus an in-place sort; laziness, short-circuiting, error propagation, parallelism all hand-rolled; the more stages, the more the loop bodies blur together
Package-level functions (pre-1.27)	Type-safe, lazy, composable	Nested calls read inside-out, fluent feel lost — the longer the pipeline, the worse
Generic methods (this library)	Top-down readability, types flow through the chain; stateful steps slot into the chain seamlessly; laziness/short-circuit/parallelism out of the box	Runtime overhead — materialization cost of stateful operators plus dispatch, quantified in Performance below

Readability is half the story; the Performance subsection below quantifies the runtime cost so you can weigh the trade-off for your workload.

Performance

Same pipelines as the style comparison above, each against its hand-written equivalent (BenchmarkTopKVsManual / BenchmarkPipelineVsManual). Both sides play by the same rules: the hand-written version collects into a fresh slice and sorts it in place (unstable pdqsort, same contract as Sorted — the source is never mutated).

Top-K (stateful: Sorted+Limit):

Scale	Pipeline	Hand-written for	Overhead
1e2	~3.5 μs	~1.5 μs	2.3x
1e4	~0.17 ms	~0.16 ms	1.1x
1e6	~17 ms	~7.0 ms	2.4x

Stateless only (Filter+Map+ToSlice):

Scale	Pipeline	Hand-written for	Overhead
1e2	~2.6 μs	~0.6 μs	4.6x
1e4	~0.29 ms	~0.17 ms	1.7x
1e6	~29 ms	~16 ms	1.8x

With the unstable pdqsort as the default, sorting itself becomes cheap and the engine's per-element cost shows through: the remaining gap is dispatch through the sink chain (interface calls + closures, roughly fixed nanoseconds per element), plus per-evaluation setup (~25 small allocations) that dominates at tiny scales. The materialization buffer is a fresh exclusive slice — Sorted/Reverse transform it in place, no extra copy. If you need stable ordering, StableSorted pays the stable-sort cost on both sides (comparable hand-written slices.SortStableFunc code trails by only ~1.2x there, since the sort dominates).

Reproduce with go test -bench . -run '^$' -benchtime 1s (AMD Ryzen 5 7535U, median of 3 runs).

NumberStream

Mirrors Java's primitive streams (IntStream/LongStream) — not for boxing avoidance (Go generics have zero boxing), but for constraint narrowing: NumberStream[N Number] embeds Stream[N], moving the element constraint into the wrapper's own type parameter. This sidesteps the Go 1.27 rule that methods cannot constrain the receiver's existing type parameter, so element-constrained APIs become chainable methods (stream.Range(0, 100).Sum() in one line):

// Narrowed chain: range → filter → sum, no package-level detour
total := stream.Range(1, 101).
    Filter(func(v int) bool { return v%2 == 0 }).
    Sum() // 2550

// Natural-order Sorted/Distinct without comparators or key functions
stream.OfNumber(3, 1, 3, 2, 1).Distinct().Sorted().ToSlice() // [1 2 3]

// Type migration into the narrowed world (Java mapToInt style)
stream.FromSlice(words).MapToNumber(func(s string) int { return len(s) }).Avg()

// Bridging: AsNumber narrows a *Stream; AsStream escapes back
// (for Zip's other side, Chunk/Enumerate, comparator-based Sorted/Min/Max)
stream.Of("a", "b").Zip(stream.Range(1, 10).AsStream(), pair)
stream.AsNumber(stream.Of(1, 2, 3)).Contains(2) // true

Narrowed method surface: element-preserving intermediates (Filter/Peek/TakeWhile/DropWhile/Limit/Skip/Reverse), natural-order ops (Sorted()/StableSorted()/Distinct()), flags/lifecycle (Parallel/Sequential/Unordered/OnClose), and narrowed terminals (Sum()/Avg()/Min()/Max()/Contains()). Non-overridden promoted methods keep Stream semantics: type-migrating operators (Map[U]/Zip/Scan) return *Stream, value terminals (ToSlice/Count/Collect) work directly. Both bridges copy the handle and mark the source consumed — one-shot semantics, second bridge panics.

Performance note: each narrowing entry and element-preserving operator costs one extra handle allocation over the equivalent Stream chain (construction-time only, ~65ns/112B; a depth-4 pure-construction chain measures +5 allocs/+560B; evaluation hot path is identical at n=1e6 — see BenchmarkNumberStreamVsStream). For rebuild-heavy/evaluate-light workloads (tiny inputs, chains rebuilt per request), chain intermediates on *Stream first and narrow with AsNumber just before the terminal.

API Overview

Category	APIs
Construction	Of OfNonZero FromSlice FromSeq FromChannel FromMap FromFunc Generate Iterate Range RangeClosed Concat (method a.Concat(b); the package-level Concat(a, b) is a deprecated adapter, slated for removal in the next version) Empty
Stateless intermediate	Filter Map FlatMap FlatMapSeq Peek TakeWhile DropWhile
Err variants	MapErr FilterErr FlatMapErr PeekErr
Stateful intermediate	Limit Skip Sorted StableSorted DistinctBy Reverse Scan
Parallelism control	Parallel(n) Sequential() Unordered()
Package-level intermediate	Distinct Sorted (natural order) Chunk Enumerate WindowSliding TimeWindow SortedByTime / CompleteTimeBuckets (TimeWindow companions: sort buckets by time / fill empty windows)
Two-stream	Zip Join (inner) LeftJoin (left outer; right outer = call LeftJoin on the right stream) — Join/LeftJoin require a known-finite right stream (SpLimited, link-time panic otherwise: swap sides / .Limit(n) / materialize)
Lifecycle	OnClose(f) Close() Cache() (replayable factory; the package-level Cache(s) is a deprecated adapter, slated for removal in the next version)
Terminal	ForEach ForEachUntil ToSlice ToSeq Count Reduce ReduceOpt Collect First FindAny AnyMatch AllMatch NoneMatch Min Max Err
Collectors (subpackage collector)	ToSlice ToSet ToMap ToMapMerge GroupingBy GroupingByDownstream PartitioningBy PartitioningBySlice Teeing Filtering FlatMapping CollectingAndThen MinBy MaxBy Joining Counting Reducing Mapping Summing Averaging Summarizing (SummaryStats)
Numeric constraints	stream.Integer/stream.Float/stream.Number (aliases of constraints subpackage)
Package-level aggregation	Sum Avg Summary Contains Min Max
Number stream (Task 18)	NumberStream[N] (embeds Stream[N]) + narrowing entries Range (returns *NumberStream) OfNumber FromNumberSlice MapToNumber AsNumber/AsStream; narrowed methods Sum() Avg() Min() Max() Contains() Sorted() StableSorted() Distinct()

For the full reference and examples, see docs/api.md.

Comparison with Java Stream

Java	go-stream	Notes
Stream<T> (interface)	*Stream[T] (concrete struct)	In Go 1.27 interface methods cannot declare type parameters; generic methods must live on concrete types
stream.of(...) / Arrays.stream	stream.Of(...) / stream.FromSlice
Collectors.toList()	collector.ToSlice[T]()
Collectors.toMap	collector.ToMap / ToMapMerge	Key conflicts: last-wins (aligned with Go map conventions); use ToMapMerge for custom merging
Collectors.groupingBy	collector.GroupingBy	Preserves encounter order within groups
Comparator	func(a, b T) int	Aligned with the standard library's slices.SortFunc/cmp.Compare conventions
IntStream specializations	NumberStream[N] narrowing wrapper + Number/cmp.Ordered constraints	Boxing avoidance is unnecessary (Go generics have zero boxing); the constraint narrowing value is kept: element-constrained APIs (Sum()/Avg()/Min()/Max()/Contains()/natural-order Sorted()/Distinct()) become chainable methods on NumberStream, mirroring Java's primitive-stream ergonomics
stream.sorted()	Sorted (unstable pdqsort) / StableSorted	Java's sorted() is always stable; go-stream defaults to the faster unstable sort (aligned with slices.SortFunc) and offers StableSorted when encounter-order preservation matters (aligned with slices.SortStableFunc)
stream.parallel()	Parallel(n) / Sequential()	TrySplit splitting + goroutines; automatically falls back to sequential after short-circuit terminals or materializing operators
stream.unordered()	Unordered()	Clears the SpOrdered flag; under parallelism, shard results are pushed as they complete (streaming merge)
stream.onClose(f) / close()	OnClose(f) / Close()	Triggered automatically at the end of evaluation (including short-circuit/error/panic paths); explicit close is idempotent; callback errors are queryable via Err()
stream.iterator()	ToSeq() iter.Seq[T]	Outbound adaptation to Go 1.23 range-over-func; consumer break short-circuits the source
Collectors.teeing	collector.Teeing	One traversal feeds two downstream collectors, then merges both results
Collectors.groupingBy(classifier, downstream)	collector.GroupingByDownstream	Two-level reduction: group first, then collect each group with a downstream collector (combiner-supported for parallel)
Gatherers.windowSliding(n)	WindowSliding(s, n)	Full windows only; fewer than n elements produce no output; package-level due to Go 1.27 instantiation-cycle limitation
Julia/Scala time windows (Akka groupedWithin, Spark tumbling window)	TimeWindow(s, ts, d)	Buckets keyed by ts(v).Truncate(d) (aligned tumbling windows, GroupBy semantics: first-seen bucket order, encounter order within bucket, late elements join their bucket, no empty buckets); per-bucket aggregation composed via .Map(...); package-level (returning Stream[TimeBucket[T]], a derived type of T, triggers the same instantiation cycle)
summaryStatistics()	Summary/Summarizing (SummaryStats[N])	Single-pass count/sum/min/max, Avg() derived without a second pass
rangeClosed(a, b) / Stream.ofNullable	RangeClosed(a, b) / OfNonZero(xs...)	Closed interval; skip zero-value elements (zero covers nil, aligned with cmp.Or terminology)
Exception propagation	Errors as values (Err()/MapErr family)	Aligned with Go's official error style
stream.distinct()	DistinctBy[K comparable](key) method / Distinct package-level	A method's own type parameters may carry the comparable constraint (keys are compile-time comparable, zero boxing); Distinct constrains the element T itself, and methods cannot constrain the receiver's T, so it stays package-level

Design Highlights

• Sink push chain: at evaluation time, sinks are wrapped in reverse starting from the terminal operation (Accept(t) bool merges Java's cancellationRequested); the data source pushes elements through the entire chain in a single pass
• Segmented evaluation: stateful operators such as Sorted/Skip first drive upstream to materialize []T, then transform and replay; Limit supports short-circuit collection from infinite sources; Skip(0) returns the original stream as a true no-op (no materialization, flags passthrough)
• Flag propagation: SpSized/SpOrdered/SpSorted/SpDistinct propagate along the pipeline (e.g. Map preserves Sized 1:1 so downstream can preallocate), informing parallel splitting decisions
• Error model: modeled after the bufio.Scanner.Err() convention — on error, terminal operations return the accumulated partial results, and Err() returns the first error

See docs/design.md for architecture details.

Roadmap

The project is in the v0.x stage: the API is not yet stable and no compatibility is promised — new features are the priority, but breaking changes may still land between minor releases. Stability guarantees begin with v1.

• v0.1 (tag v0.1.0): sequential evaluation engine, full operator set, Collector system, errors-as-values model; Parallel(n)/Sequential() parallel evaluation; OnClose(f)/Close() lifecycle, replayable Cache, Unordered() streaming merge
• v0.2 (tag v0.2.0): Java 25 parity — ToSeq() range-over-func interop, collector composition ecosystem, WindowSliding, single-pass statistics, RangeClosed/OfNonZero; NumberStream[N] numeric narrowing; Sorted/StableSorted sort semantics split
• v0.3 (tag v0.3.0): Join/LeftJoin two-stream conditional joins (SpLimited link-time finiteness guard); TimeWindow time-window bucketing with SortedByTime/CompleteTimeBuckets companions; flag propagation aligned to Java PRESERVE semantics; panic messages unified to English
• v0.4: scope TBD — the next batch will be scoped from real-world feedback on the v0.3 API surface; suggestions welcome via issues

License

MIT