stream

A Go stream processing library that brings Java Streams-like functional operations to Go collections using generics and iter.Seq.

Requires Go 1.27+ (see go.mod). Existing Streamer source code remains compatible; the minimum compiler version increases from Go 1.26.

Features

• True lazy evaluation — intermediate operations compose iter.Seq[T] closures; nothing runs until a terminal operation iterates
• Short-circuiting — First(), AnyMatch(), Limit() stop processing as soon as the result is known
• Generics — fluent generic methods on Pipeline[T], with the legacy Streamer[T] API retained
• iter.Seq integration — Seq() method and From/From2 factory functions for native for range interop
• Parallel processing — concurrent execution via goroutine worker pools
• Functional pipelines — filter, map, flatmap, reduce, sort, distinct, and more
• Infinite streams — supplier-based streams for generator patterns

Installation

go get github.com/tr1v3r/stream

Quick Start

package main

import (
    "fmt"
    "github.com/tr1v3r/stream"
)

func main() {
    // Filter odd numbers, square them, sum the result
    sum := stream.SliceOf(1, 2, 3, 4, 5).
        Filter(func(n int) bool { return n%2 == 1 }).
        Map(func(n int) int { return n * n }).
        Reduce(func(a, b int) int { return a + b })
    fmt.Println(sum) // 35
}

Concrete pipelines and generic methods

Use Of(...) for a concrete Pipeline[T]. Type-changing operations remain in one fluent chain, with no any result or type assertion:

names := stream.Of(1, 2, 2, 3).
    Filter(func(n int) bool { return n > 1 }).
    MapTo(strconv.Itoa).
    DistinctBy(func(s string) string { return s }).
    ToSlice() // []string{"2", "3"}

text := stream.Of(1, 2).
    FlatMapTo(func(n int) stream.Pipeline[string] {
        return stream.Of(strconv.Itoa(n), "!")
    }).
    Fold("", func(result, item string) string { return result + item }) // "1!2!"

count := stream.Of(1, 2, 3).
    CollectTo(func(values ...int) int { return len(values) }) // int

Generic method	Result	Execution
MapTo[R](func(T) R)	Pipeline[R]	Lazy serial type change; exact size preserved
FlatMapTo[R](func(T) Pipeline[R])	Pipeline[R]	Lazy child concatenation; size unknown
DistinctBy[K comparable](func(T) K)	Pipeline[T]	Serial exact-key deduplication
Fold[R](initial R, func(R, T) R)	R	Incremental reduction; partial result on cancellation
CollectTo[R](func(...T) R)	R	Materializes input, then calls the collector once

Type arguments are normally inferred. Every intermediate operation on Pipeline returns a concrete pipeline, including Filter, Map, Sort, Limit, Parallel, and Execute. Same-type Map retains fused parallel execution. MapTo and FlatMapTo close the current section, execute their type-changing work serially, and retain context, worker count, and Ordered preference for downstream operations. Children of FlatMapTo may have their own parallel sections.

Compatibility and migration

Old factories (SliceOf, From, RepeatN, etc.), the Streamer interface, package-level MapTo / DistinctBy, and callback types keep their signatures. Use Wrap to enter the concrete API and .Stream() to return to an existing interface-based call site:

legacy := stream.RepeatN(3, 2)
typed := stream.Wrap(legacy).MapTo(strconv.Itoa)
var compatible stream.Streamer[string] = typed.Stream()

Pipeline does not implement Streamer: Go has neither covariant method return types nor generic interface methods. Both APIs share the same execution engine. Wrapping native streams preserves their options; foreign Streamer implementations are adapted through Seq, with unknown size and serial downstream defaults. Adapters do not copy or consume the source, so they do not enable multiple terminal operations on a single-use stream. A zero-value Pipeline is empty; Execute remains the explicit re-iterable snapshot operation.

The legacy operation tables below still describe Streamer. Pipeline exposes the same operations with concrete intermediate results, plus the five methods above. See design and performance comparison.

Stream Creation

Function	Description
SliceOf[T](slice ...T)	Create a stream from a slice or variadic elements
From[T](seq, sizeHint)	Create from an iter.Seq[T] (supports infinite streams)
From2[K, V](seq)	Create from an iter.Seq2[K, V]
Repeat[T](t T)	Create an infinite stream repeating t
RepeatN[T](t T, n int64)	Create a stream repeating t exactly n times
Concat[T](dst, ...src)	Concatenate multiple streams

// From an iter.Seq
fib := stream.From(func(yield func(int) bool) {
    a, b := 0, 1
    for yield(a) { a, b = b, a+b }
}, -1).Limit(10)

// Repeat
fives := stream.RepeatN(5, 10) // [5, 5, 5, 5, 5, 5, 5, 5, 5, 5]

Intermediate Operations

All intermediate operations are lazy — they compose closures without processing elements.

Stateless

Method	Signature	Description
Filter	(Judge[T]) Streamer[T]	Keep elements matching the predicate
Map	(Mapper[T]) Streamer[T]	Transform each element (same type)
Convert	(Converter[T, any]) Streamer[any]	Transform to a different type. Deprecated: use stream.MapTo
Peek	(Consumer[T]) Streamer[T]	Apply an action without modifying elements
FlatMap	(func(T) Streamer[any]) Streamer[any]	Flatten each element to a sub-stream

Package-level generic (interface methods cannot add type parameters): stream.MapTo[T, R](s, func(T) R) Streamer[R] — the type-safe replacement for Convert.

When to use MapTo vs Convert

MapTo keeps the result type at compile time — no Streamer[any] round-trip, no Collect(AnyTo[T]()) assertion that can panic at runtime. The trade-off: as a function it interrupts method chaining at the type-changing point, while Convert chains fluently but erases types.

// MapTo: type-safe, result is Streamer[string] — recommended default
names := stream.MapTo(stream.SliceOf(1, 2, 3), func(n int) string {
    return fmt.Sprintf("#%d", n)
})

// Existing Streamer code can adopt the concrete API at any point
stream.Wrap(stream.SliceOf(1, 2, 3)).
    Filter(func(n int) bool { return n > 1 }).
    MapTo(strconv.Itoa).
    ToSlice() // []string{"2", "3"}

// Mid-pipeline type change in a long chain: Convert keeps it readable,
// at the cost of any + a runtime assertion to come back
stream.SliceOf(1, 2, 3, 4).
    Filter(func(n int) bool { return n > 2 }).
    Convert(func(n int) any { return float64(n) * 1.5 }).
    Map(func(x any) any { return x }).              // still Streamer[any] down here
    Collect(stream.AnyTo[float64]()).([]float64)     // runtime type assertion

Rule of thumb: prefer MapTo (type change at the pipeline head, or safety matters more than fluency); Convert remains valid for mid-chain type changes in throwaway code — it is deprecated, not removed, and still works. On the concrete Pipeline API, use the generic MapTo method for both type safety and fluent chaining.

stream.SliceOf(1, 2, 3, 4).
    Filter(func(n int) bool { return n > 2 }).   // [3, 4]
    Map(func(n int) int { return n * 10 }).       // [30, 40]
    Peek(func(n int) { fmt.Println(n) })          // prints 30, 40

// FlatMap
stream.SliceOf(1, 2, 3).
    FlatMap(func(n int) stream.Streamer[any] {
        return stream.SliceOf[any](n, n*10)
    }) // [1, 10, 2, 20, 3, 30]

Stateful

Method	Signature	Description
Distinct	() Streamer[T]	Remove duplicate elements
DistinctBy	(Streamer[T], func(T) K) Streamer[T]	Dedup by comparable key (no string coercion)
Sort	(Comparator[T]) Streamer[T]	Sort ascending
ReverseSort	(Comparator[T]) Streamer[T]	Sort descending
Reverse	() Streamer[T]	Reverse element order
Limit	(int64) Streamer[T]	Take at most N elements; an unknown input count stays unknown for positive N
Skip	(int64) Streamer[T]	Skip first N elements
Pick	(start, end, interval int) Streamer[T]	Pick elements at intervals

stream.SliceOf(3, 1, 4, 1, 5).
    Distinct().                                    // [3, 1, 4, 5]
    Sort(func(a, b int) int { return a - b }).     // [1, 3, 4, 5]
    Limit(2)                                       // [1, 3]

// Dedup with exact comparable keys (5x faster, 300x fewer allocs than Distinct)
byDept := stream.DistinctBy(users, func(u User) string { return u.Dept })

Terminal Operations

Collecting

Method	Signature	Description
ToSlice	() []T	Collect all elements into a slice
Collect	(Collector[T]) any	Collect using a custom collector
ForEach	(Consumer[T])	Iterate over each element
Count	() int64	Return the number of elements

Reduce

Method	Signature	Description
Reduce	(BinaryOperator[T]) T	Reduce with zero-value init
ReduceFrom	(T, BinaryOperator[T]) T	Reduce with explicit init value
ReduceWith	(any, Accumulator[T, any]) any	Reduce with different accumulator type
ReduceBy	(initBuilder, Accumulator[T, any]) any	Reduce with size-aware init builder

Match

Method	Signature	Description
AllMatch	(Judge[T]) bool	True if all elements match
NonMatch	(Judge[T]) bool	True if no elements match
AnyMatch	(Judge[T]) bool	True if any element matches

Element

Method	Signature	Description
First	() T	First element
Take	() T	Random element (uniform reservoir sampling, O(1) memory)
Any	() T	Alias for Take
Last	() T	Last element

iter.Seq Integration

// Convert a stream to iter.Seq for native range loops
for v := range stream.SliceOf(1, 2, 3).Filter(func(n int) bool { return n > 1 }).Seq() {
    fmt.Println(v) // 2, 3
}

// Create a stream from an existing iter.Seq
seq := slices.Values([]int{10, 20, 30})
stream.From(seq, 3).Map(func(n int) int { return n * 2 }).ToSlice() // [20, 40, 60]

// Create a stream from iter.Seq2 (uses values only)
m := map[string]int{"a": 1, "b": 2}
stream.From2(maps.All(m)).ToSlice() // [1, 2] (order varies)

Parallel Processing

Parallelism is section-scoped: Parallel(n) opens a section of stateless operations that run fused on ONE worker pool — not one pool per operation. Consecutive Filter/Map/Peek inside the section compose into a single function; elements flow in 64-element batches.

// one pool executes the whole Filter+Map section (4 workers)
stream.SliceOf(largeData...).
    Parallel(4).                        // open section, 4 workers
    Filter(heavyPredicate).
    Map(heavyTransform).
    ForEach(process)

Sections close at stateful ops (Sort, Distinct, Limit, ...), type changes (MapTo, Convert, FlatMap), and every terminal. A mid-chain Parallel(n) closes the current section and opens a new one — this is how you size concurrency per cost profile while sections overlap (pipeline parallelism):

// heterogeneous: 16 workers absorb IO latency, 2 suffice for CPU parsing,
// and section A keeps producing while section B consumes
stream.SliceOf(urls...).
    Parallel(16).                       // section A: IO-bound
    Filter(func(u string) bool { return checkRemote(u) }).
    Parallel(2).                        // closes A, opens section B
    Map(func(u string) string { return parse(u) })

Order: sections are unordered by default (fastest). Add Ordered() to reproduce serial encounter order exactly:

stream.SliceOf(data...).Parallel(4).Ordered().
    Filter(f).Map(g).ToSlice()          // element-for-element equal to serial

Parallel(n) behavior:

• n <= 0: synchronous (no change)
• n >= 1: n workers on the section's fused stages, unordered
• Ordered(): same, but output order matches serial execution
• Sections ignore Parallel for stateful stages (they materialize serially after the section closes)

Overhead (measured on near-free workloads, Apple M3 Pro): unordered sections run at ~1–2× serial time, ordered ~2×; heavy per-element work scales at ~3.6× with 4 workers. See docs/proposals/parallel-v2.md.

Ordered execution limits outstanding work to n batches (64 elements each). A slow early batch applies backpressure to the source instead of retaining an unbounded backlog of later results.

Short-circuiting a parallel section cancels its upstream execution, including previous parallel sections and internal Concat/FlatMap sources. Cancellation is local to that execution: it does not cancel the caller's context or prevent subsequent stages such as Limit(1).Append(...) from running. Workers check cancellation before each element; callbacks already running must cooperate.

Use WithContext(ctx) to support cancellation:

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
stream.SliceOf(data...).WithContext(ctx).Parallel(4).ForEach(work)

Helper Functions

// To converts []T to []R
floats := stream.To(func(n int) float64 { return float64(n) })(1, 2, 3).([]float64)

// AnyTo converts []any to []T
items := stream.AnyTo[int]()(1, 2, 3).([]int)

Type Definitions

The types package defines functional interfaces as function types:

type Judge[T any] func(T) bool                    // Predicate
type Mapper[T any] func(T) T                      // Same-type transform
type Converter[T, R any] func(T) R                // Type transform
type Comparator[T any] func(T, T) int             // Ordering
type Consumer[T any] func(T)                      // Side-effect action
type BinaryOperator[T any] func(T, T) T           // Same-type accumulator
type Accumulator[T, R any] func(R, T) R           // Cross-type accumulator
type Collector[T any] func(...T) any              // Collect to result
type Unique interface{ Key() string }             // Custom distinct key

Important Notes

• Infinite streams hang non-short-circuiting terminals. ToSlice, ForEach, Reduce*, Count, Last, and Take/Any (without a cancellable context) never finish on Repeat or an infinite From source. Bound them with Limit or WithContext — cancellation is checked at element boundaries, so every terminal returns promptly once the context is cancelled (collections come back empty, reductions keep their partial result):

stream.Repeat(1).Limit(100).ToSlice() // bounded: ok

ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
stream.Repeat(1).WithContext(ctx).Take() // cancellable: ok

• Streams are single-use. A terminal operation consumes the stream. Create a new stream for each pipeline.
• Lazy evaluation — intermediate operations compose closures; work happens only during terminal operations. Limit(1).First() on a million elements only processes one element.
• Distinct uses fmt.Sprint by default for hashing. Implement the types.Unique interface (Key() string) for custom hash keys, or use the generic stream.DistinctBy with comparable keys for exact equality without string coercion.
• Parallel sections are unordered by default. Use Parallel(n).Ordered() when output must match serial order, or Sort after the section.

License

MIT