stream

stream

English | 日本語 | 中文 | 한국어 | Español | Português

A Go generic stream processing library. Chainable collection operations for filter, map, sort, group, and more — with lazy evaluation by default. Try it online!

All operations are lazy. Operations that require full data (Sort, Reverse, Shuffle, TakeLast, Chunk, Partition) buffer internally and resume lazy iteration automatically.

Requirements

• Go 1.23 or later (uses iter.Seq[T] and range-over-function)

Install

go get github.com/nd-forge/stream

Quick Start

import "github.com/nd-forge/stream"

// Method chaining for same-type operations
result := stream.Of(5, 2, 8, 1, 9, 4, 7, 3, 6).
    Filter(func(n int) bool { return n%2 == 0 }).
    Sort(func(a, b int) int { return b - a }).
    Take(3).
    ToSlice()
// [8, 6, 4]

// Top-level functions for type-changing operations
names := stream.Map(
    stream.Of(users...).Filter(func(u User) bool { return u.IsActive }),
    func(u User) string { return u.Name },
).ToSlice()

Design

Lazy by Default

All operations build a lazy pipeline internally using iter.Seq[T]. No intermediate slices are allocated until a terminal operation (ToSlice, ForEach, Reduce, etc.) is called.

Operations that inherently need all data — Sort, Reverse, Shuffle, TakeLast, Chunk, Partition — buffer internally, then resume lazy iteration for subsequent operations.

Type Parameters

Go does not allow methods to introduce new type parameters. This library separates:

API

Constructors

Generators (Infinite Sequences)

Chainable Methods

Operations that return Stream[T] and can be chained.

Sort, Reverse, Shuffle, TakeLast buffer all elements internally.

Terminal Operations

Transform Functions

Top-level functions for type-changing operations.

Numeric Functions

Specialized operations for numeric streams (int, float64, etc.).

iter.Seq Bridge

Examples

Types used in examples below:

type Product struct {
    Name     string
    Category string
    Price    float64
    InStock  bool
}

type User struct {
    Name     string
    Age      int
    IsActive bool
    Orders   []Order
}

type Order struct {
    Product  string
    Amount   float64
    Discount float64
}

Filter, Sort, Take [playground]

products := stream.Of(
    Product{Name: "Laptop", Category: "Electronics", Price: 1200, InStock: true},
    Product{Name: "T-Shirt", Category: "Clothing", Price: 25, InStock: true},
    Product{Name: "Headphones", Category: "Electronics", Price: 150, InStock: false},
    Product{Name: "Jeans", Category: "Clothing", Price: 80, InStock: true},
)

// In-stock products, sorted by price descending, top 3
top3 := products.
    Filter(func(p Product) bool { return p.InStock }).
    Sort(func(a, b Product) int {
        if a.Price > b.Price { return -1 }
        if a.Price < b.Price { return 1 }
        return 0
    }).
    Take(3).
    ToSlice()

Map and FlatMap [playground]

// Extract product names
names := stream.Map(
    products.Filter(func(p Product) bool { return p.InStock }),
    func(p Product) string { return p.Name },
).ToSlice()

// Flatten nested orders from users
allOrders := stream.FlatMap(
    stream.Of(users...),
    func(u User) []Order { return u.Orders },
)

GroupBy and Aggregate [playground]

byCategory := stream.GroupBy(products, func(p Product) string { return p.Category })

for category, group := range byCategory {
    total := stream.SumBy(stream.Of(group...), func(p Product) float64 { return p.Price })
    fmt.Printf("%s: total=$%.2f count=%d\n", category, total, len(group))
}

Partition and Chunk [playground]

// Split by condition
inStock, outOfStock := products.Partition(func(p Product) bool { return p.InStock })

// Batch processing
batches := stream.From(items).Chunk(100)
for _, batch := range batches {
    api.Send(batch.ToSlice())
}

Zip [playground]

names := stream.Of("Alice", "Bob", "Charlie")
scores := stream.Of(85.0, 92.0, 78.0)

pairs := stream.Zip(names, scores).ToSlice()
// [{Alice 85}, {Bob 92}, {Charlie 78}]

Infinite Sequences [playground]

// First 5 even natural numbers
evens := stream.Naturals().
    Filter(func(n int) bool { return n%2 == 0 }).
    Take(5).
    ToSlice()
// [0, 2, 4, 6, 8]

// Fibonacci sequence
fib := stream.Map(
    stream.Iterate(
        stream.Pair[int, int]{First: 0, Second: 1},
        func(p stream.Pair[int, int]) stream.Pair[int, int] {
            return stream.Pair[int, int]{First: p.Second, Second: p.First + p.Second}
        },
    ).Take(10),
    func(p stream.Pair[int, int]) int { return p.First },
).ToSlice()
// [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

// Lazy evaluation: processes only 2,001 of 1,000,000 elements
result := stream.Range(0, 1_000_000).
    Filter(func(n int) bool { return n%1000 == 0 }).
    Take(3).
    ToSlice()
// [0, 1000, 2000]

iter.Seq Bridge [playground]

// Standard library interop
keys := stream.Collect(maps.Keys(myMap)).Sort(cmp).ToSlice()

// for-range support
for v := range stream.Of(1, 2, 3).Seq() {
    fmt.Println(v)
}

Benchmark

10,000 int elements — Apple M1. Compared with samber/lo.

Filter + Take (lazy evaluation advantage)

Benchmark                    ns/op       B/op    allocs/op
────────────────────────────────────────────────────────────
Native                         111        248          5
Stream                         346        528         15   ← 3.1x native
lo                          25,000     81,920          1   ← 72x slower than Stream

Stream's lazy evaluation short-circuits — only processes elements until Take is satisfied. lo must filter all 10,000 elements first, then take.

Chained: Filter → Map → Take 5

Benchmark                    ns/op       B/op    allocs/op
────────────────────────────────────────────────────────────
Native                         171        260          9
Stream                         384        544         19   ← 2.2x native
lo                          64,600    152,601      3,336   ← 168x slower than Stream

Full scan (no early termination)

Benchmark                    ns/op       B/op    allocs/op
────────────────────────────────────────────────────────────
Native Filter              19,400    128,249         16
lo     Filter              25,600     81,920          1
Stream Filter              42,200    128,409         22

Native Reduce               3,300          0          0
lo     Reduce               9,700          0          0
Stream Reduce              23,400         64          2

For full scans without early termination, lo is faster than Stream due to lower abstraction overhead. Stream's advantage shines in pipelines with Take, First, Find, or other short-circuiting operations.

Run cd _benchmark && go test -bench=. -benchmem ./... to reproduce.

License

MIT