stream

package module
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Aug 23, 2026 License: MIT Imports: 7 Imported by: 0

README

stream

CI Go Version Go Reference

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

Requires Go 1.26+ (see go.mod).

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 — type-safe streams with Streamer[T]
  • 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
}

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 (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)
})

// Head-of-pipeline type change: MapTo costs nothing — chain continues below it
stream.MapTo(stream.SliceOf(employees...), func(e *Employee) Dept { return e.Dept }).
    Filter(func(d Dept) bool { return d.Active }).  // normal chaining resumes
    Map(func(d Dept) string { return d.Name })

// 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. When Go ships generic methods, a Map[R](func(T) R) Streamer[R] method can offer both.

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

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:
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

Documentation

Overview

Package stream provides Java Streams-like functional operations on Go collections.

It enables true lazy evaluation, parallel processing, and functional-style pipelines using Go generics and the iter package (requires Go 1.26+, per go.mod).

Intermediate operations compose iter.Seq[T] closures without executing any work. Processing is deferred until a terminal operation ranges over the pipeline. Short-circuit operations (First, AnyMatch, Limit) naturally stop early.

Quick Start

Create a stream from a slice, apply intermediate operations, and collect results: 

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 })
// sum == 35

Stream Creation

Use factory functions to create streams:

  • SliceOf: from a slice or variadic elements
  • From: from an iter.Seq[T] (supports infinite streams)
  • From2: from an iter.Seq2[K, V]
  • Repeat: infinite repeating element
  • RepeatN: element repeated N times
  • Concat: combine multiple streams

Operations

Intermediate (lazy, return a new stream):

  • Stateless: Filter, Map, Convert (deprecated: use generic MapTo), Peek, FlatMap
  • Stateful: Distinct (or generic DistinctBy), Sort, ReverseSort, Reverse, Limit, Skip, Pick

Terminal (eager, execute the pipeline):

  • Collect: ToSlice, Collect
  • Iterate: ForEach, Seq (native iter.Seq[T] for range loops)
  • Reduce: Reduce, ReduceFrom, ReduceWith, ReduceBy
  • Match: AllMatch, NonMatch, AnyMatch
  • Element: First, Take, Any, Last
  • Count: Count

iter.Seq Integration

The Seq() method returns the underlying iter.Seq[T] for use with Go's range: 

for v := range stream.SliceOf(1, 2, 3).Seq() {
    fmt.Println(v)
}

Parallel Processing

Parallel(n) opens a section of stateless operations that run fused on one worker pool; consecutive Filter/Map/Peek inside the section compose into a single stage. Sections close at stateful operations, type changes, every terminal, and the next Parallel call (which opens a new section — size concurrency per cost profile while sections overlap). Output is unordered unless Ordered() follows: 

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

See docs/proposals/parallel-v2.md for semantics and measured overheads.

Helper Functions

  • To[T, R]: converts a slice of T to a slice of R via a converter
  • AnyTo[T]: converts []any to []T via type assertion

Important

Streams are single-use. Each terminal operation consumes the underlying iterator. Create a new stream for each pipeline.

Infinite sources (Repeat, unbounded From) hang non-short-circuiting terminal operations such as ToSlice, Reduce, Count, Last, or Take without a cancellable context — bound them with Limit or WithContext.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AnyTo 

func AnyTo[T any](data ...any) types.Collector[any]

AnyTo converts a slice of any to a slice of T

func To 

func To[T, R any](converter types.Converter[T, R]) types.Collector[T]

To converts a slice of T to a slice of R

Types

type Streamer 

type Streamer[T any] interface {
	// WithContext sets the context consulted by later operations; a
	// cancelled context makes intermediate operations stop pulling and
	// terminals return promptly. Applies to the returned stream only.
	WithContext(context.Context) Streamer[T]

	// Filter keeps elements for which judge returns true. Lazy, order
	// preserving; sizeHint becomes unknown (-1).
	Filter(types.Judge[T]) Streamer[T]
	// Map transforms each element with a same-type function m.
	// For a different result type use the generic stream.MapTo.
	Map(types.Mapper[T]) Streamer[T]
	// Convert transforms elements to any. Deprecated: it loses the element
	// type and forces type assertions downstream; use the generic
	// stream.MapTo[T, R] instead. Kept for backward compatibility.
	//
	// Deprecated: use MapTo.
	Convert(types.Converter[T, any]) Streamer[any]
	// Peek applies consumer to each element as it passes through without
	// changing them; useful for debugging or side effects mid-pipeline.
	Peek(types.Consumer[T]) Streamer[T]
	// FlatMap flattens each element to a sub-stream and concatenates
	// sub-streams' elements. The result is Streamer[any]; use AnyTo or To
	// to recover concrete types. sizeHint becomes unknown.
	FlatMap(func(T) Streamer[any]) Streamer[any]

	// Distinct removes duplicate elements keeping first occurrences. Keys
	// come from fmt.Sprint (or types.Unique), so 1 and "1" collide —
	// prefer the generic stream.DistinctBy for exact comparable keys.
	Distinct() Streamer[T]
	// Sort orders elements ascending by comparator (slices.SortFunc).
	// Materializes the pipeline stage when iterated; sizeHint preserved.
	Sort(types.Comparator[T]) Streamer[T]
	// ReverseSort orders elements descending by comparator.
	ReverseSort(types.Comparator[T]) Streamer[T]
	// Reverse yields elements in reverse order; empty input stays empty.
	Reverse() Streamer[T]
	// Limit keeps at most the first n elements (n <= 0 yields empty).
	// Short-circuits the upstream pipeline.
	Limit(int64) Streamer[T]
	// Skip discards the first n elements (n <= 0 keeps everything).
	Skip(int64) Streamer[T]
	// Pick selects elements at absolute indices start, start+interval, ...
	// up to end inclusive; end < 0 means the last index (materializing the
	// stage when sizeHint is unknown); start < 0 or interval <= 0 yields
	// empty.
	Pick(startIndex, endIndex, interval int) Streamer[T]

	// Append yields the stream's elements followed by data. sizeHint grows
	// by len(data) when known.
	Append(...T) Streamer[T]
	// Execute eagerly materializes the pipeline so far and returns a
	// re-iterable snapshot stream; ctx and parallelSize carry over.
	Execute() Streamer[T]

	// Parallel sets worker-pool concurrency for a section of stateless
	// operations: n <= 0 keeps synchronous execution, n >= 1 runs n
	// workers on the section's fused stages (Filter/Map/Peek chain into
	// one pool). A mid-chain call closes the current section and opens a
	// new one; stateful ops and type changes close sections too. Order is
	// not preserved unless Ordered() follows. See
	// docs/proposals/parallel-v2.md.
	Parallel(int) Streamer[T]
	// Ordered marks the current (or next) parallel section as
	// order-preserving: elements are index-tagged and re-sequenced at the
	// consumer, so output matches serial execution order. No-op in serial
	// mode. Costs one index stamp and slot lookup per element.
	Ordered() Streamer[T]

	// ToSlice collects all elements into a new slice; hangs on infinite
	// sources unless bounded or cancellable.
	ToSlice() []T
	// Collect drains the stream into the caller-provided collector and
	// returns its result (type any — assert it back).
	Collect(types.Collector[T]) any
	// ForEach applies consumer to every element in order.
	ForEach(types.Consumer[T])
	// AllMatch reports whether judge holds for every element; false on the
	// first violation (short-circuit).
	AllMatch(types.Judge[T]) bool
	// NonMatch reports whether judge holds for no element; false on the
	// first match (short-circuit).
	NonMatch(types.Judge[T]) bool
	// AnyMatch reports whether judge holds for at least one element; true
	// on the first match (short-circuit).
	AnyMatch(types.Judge[T]) bool
	// Reduce folds elements with accumulator, starting from T's zero
	// value; empty input returns the zero value.
	Reduce(accumulator types.BinaryOperator[T]) T
	// ReduceFrom folds elements starting from initValue.
	ReduceFrom(initValue T, accumulator types.BinaryOperator[T]) T
	// ReduceWith folds elements into an any-typed accumulator, enabling
	// cross-type reduction; assert the result back to the concrete type.
	ReduceWith(initValue any, accumulator types.Accumulator[T, any]) any
	// ReduceBy builds its initial value from the stream's sizeHint (which
	// may be negative = unknown, e.g. for capacity preallocation), then
	// folds like ReduceWith.
	ReduceBy(initValueBuilder func(sizeMayNegative int) any, accumulator types.Accumulator[T, any]) any
	// First returns the first element or T's zero value when empty;
	// short-circuits the pipeline.
	First() T
	// Take returns a uniformly random element via reservoir sampling —
	// O(1) memory, honors cancellation; zero value when empty.
	Take() T
	// Any is an alias for Take.
	Any() T
	// Last returns the final element or T's zero value; consumes the whole
	// stream and hangs on infinite sources.
	Last() T
	// Count returns the element count in O(1) when sizeHint is known,
	// otherwise by full iteration.
	Count() int64
	// Seq returns the underlying iter.Seq[T] for native range loops.
	Seq() iter.Seq[T]
}

Streamer is a lazily-evaluated pipeline over elements of type T, in the spirit of Java Streams. Intermediate operations compose iter.Seq[T] closures and do no work until a terminal operation iterates; short-circuit terminals (First, AnyMatch, Limit-fed pipelines) stop early.

Streams are single-use: a terminal operation consumes the underlying sequence. Create a new stream for each pipeline.

Usage pattern

Create (SliceOf, From, Repeat, Concat...), transform (Filter, Map...), terminate (ToSlice, Reduce, ForEach...): 

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

func Concat 

func Concat[T any](srcs ...Streamer[T]) Streamer[T]

Concat concatenates srcs in order: all elements of the first stream, then the second, and so on. sizeHint is unknown (-1) after concatenation. An empty argument list yields an empty stream; short-circuiting the result stops pulling the remaining sources.

func DistinctBy added in v0.2.0 

func DistinctBy[T any, K comparable](s Streamer[T], key func(T) K) Streamer[T]

DistinctBy removes elements whose key, produced by key, has already been seen, keeping first occurrences. Keys use Go map equality (K comparable), avoiding the string-coercion collisions of Distinct. Like Distinct it runs serially — the shared key map is not concurrency-safe — while preserving parallelSize for downstream operations.

func From added in v0.1.0 

func From[T any](seq iter.Seq[T], sizeHint int64) Streamer[T]

From wraps an existing iter.Seq[T] as a Streamer. sizeHint declares the known element count when non-negative (-1 for unknown or infinite); it feeds Count's O(1) fast path and capacity preallocation, so keep it honest. The seq is consumed once — streams are single-use. Infinite sequences are supported but must be bounded (Limit) or made cancellable (WithContext) before a non-short-circuiting terminal operation.

func From2 added in v0.1.0 

func From2[K, V any](seq iter.Seq2[K, V]) Streamer[V]

From2 adapts an iter.Seq2[K, V] (e.g. maps.All) into a Streamer of the values only; keys are discarded. sizeHint is unknown (-1). Map iteration order is unspecified, so the resulting stream order varies.

func MapTo added in v0.2.0 

func MapTo[T, R any](s Streamer[T], m types.Converter[T, R]) Streamer[R]

MapTo transforms each element of s from T to R, preserving ctx, sizeHint and parallelSize. Unlike Convert it keeps the result type, so no Streamer[any] round-trip with type assertions is needed: 

names := stream.MapTo(stream.SliceOf(1, 2, 3), func(n int) string {
    return fmt.Sprintf("#%d", n)
}).ToSlice()

As a package function it interrupts method chaining at the type-changing point; Convert chains fluently but erases the element type. Prefer MapTo, especially for head-of-pipeline type changes where chaining resumes right below it; Convert stays acceptable for mid-chain changes despite being deprecated.

func Repeat 

func Repeat[T any](t T) Streamer[T]

Repeat creates an infinite stream of the same value.

func RepeatN 

func RepeatN[T any](t T, count int64) Streamer[T]

RepeatN creates a stream repeating t exactly n times.

func SliceOf 

func SliceOf[T any](slice ...T) Streamer[T]

SliceOf creates a stream from a slice or variadic elements.

Source Files

View all Source files

Directories

Path Synopsis

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.