stream

README

stream

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

Requires Go 1.23+.

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[T](seq, sizeHint)	Create from a Go iter.Seq[T]
From2[K, V](seq)	Create from a Go iter.Seq2[K, V]

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

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

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

stream.SliceOf(largeData...).
    Parallel(4).                        // 4 concurrent workers
    Filter(heavyPredicate).
    Map(heavyTransform).
    ForEach(process)

Parallel(n) behavior:

• n <= 0: synchronous (no change)
• n >= 1: concurrent workers with goroutine pools

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

• 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.
• Parallel mode does not preserve order. Elements may be processed out of order when using Parallel(n) with n > 1. Use Sort after parallel operations if order matters.

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.23+).

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, Peek, FlatMap
• Stateful: Distinct, 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

Use Parallel(n) to enable concurrent processing. n controls concurrency:

• 0: no change (synchronous)

• 1+: concurrent workers

  stream.SliceOf(data...).Parallel(4).Filter(...).ForEach(...)

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.

Index

• Variables
• func AnyTo[T any](data ...any) types.Collector[any]
• func To[T, R any](converter types.Converter[T, R]) types.Collector[T]
• type Sortable
  • func (a *Sortable[T]) Len() int
  • func (a *Sortable[T]) Less(i, j int) bool
  • func (a *Sortable[T]) Swap(i, j int)
• type Streamer
  • func Concat[T any](srcs ...Streamer[T]) Streamer[T]
  • func From[T any](seq iter.Seq[T], sizeHint int64) Streamer[T]
  • func From2[K, V any](seq iter.Seq2[K, V]) Streamer[V]
  • func Repeat[T any](t T) Streamer[T]
  • func RepeatN[T any](t T, count int64) Streamer[T]
  • func SliceOf[T any](slice ...T) Streamer[T]

Variables

var (
	// ErrUnsupportType unsupport type
	ErrUnsupportType = errors.New("unsupport type")
)

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 Sortable

type Sortable[T any] struct {
	List []T
	Cmp  types.Comparator[T]
}

func (*Sortable[T]) Len

func (a *Sortable[T]) Len() int

func (*Sortable[T]) Less

func (a *Sortable[T]) Less(i, j int) bool

func (*Sortable[T]) Swap

func (a *Sortable[T]) Swap(i, j int)

type Streamer

type Streamer[T any] interface {
	// WithContext set Streamer context
	WithContext(context.Context) Streamer[T]

	// Filter filter data by Judge result
	Filter(types.Judge[T]) Streamer[T]
	Map(types.Mapper[T]) Streamer[T]
	Convert(types.Converter[T, any]) Streamer[any]
	Peek(types.Consumer[T]) Streamer[T]
	// FlatMap flattens each element to a sub-stream and concatenates
	FlatMap(func(T) Streamer[any]) Streamer[any]

	Distinct() Streamer[T]
	Sort(types.Comparator[T]) Streamer[T]
	ReverseSort(types.Comparator[T]) Streamer[T]
	Reverse() Streamer[T]
	Limit(int64) Streamer[T]
	Skip(int64) Streamer[T]
	Pick(startIndex, endIndex, interval int) Streamer[T]

	// Append append data to streamer source
	Append(...T) Streamer[T]
	// Execute eager execute streamer stage
	Execute() Streamer[T]

	// Parallel 0 do nothing, 1 async work, 2-n concurrent work
	Parallel(int) Streamer[T]

	ToSlice() []T
	Collect(types.Collector[T]) any
	ForEach(types.Consumer[T])
	// Match methods
	AllMatch(types.Judge[T]) bool
	NonMatch(types.Judge[T]) bool
	AnyMatch(types.Judge[T]) bool
	// Reduce reduce calculate
	Reduce(accumulator types.BinaryOperator[T]) T
	ReduceFrom(initValue T, accumulator types.BinaryOperator[T]) T
	ReduceWith(initValue any, accumulator types.Accumulator[T, any]) any
	ReduceBy(initValueBuilder func(sizeMayNegative int) any, accumulator types.Accumulator[T, any]) any
	// Pick one
	First() T
	Take() T
	Any() T
	Last() T
	// Count return count result
	Count() int64
	// Seq returns the underlying iter.Seq[T] for native range loops
	Seq() iter.Seq[T]
}

func Concat

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

func From

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

func From2

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

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.