streams

package module
v2.0.2 Latest Latest
Warning

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

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

README

go-streams

Go Reference Go Report Card Build Status codecov

Lazy, composable sequences for Go 1.27, built directly on iter.Seq.

go get github.com/coldsmirk/go-streams/v2

import streams "github.com/coldsmirk/go-streams/v2"

names := streams.Of(users...).
    Filter(func(u User) bool { return u.Active }).
    Map(func(u User) string { return u.Name }).
    Map(strings.ToUpper).
    Collect()

That chain changes the element type twice without leaving the chain. In v1 it could not: Go had no type parameters on methods, so every type-changing operation had to be a free function and the pipeline turned inside out. Go 1.27 added generic methods, and v2 is the redesign that follows from them.

Migrating from v1: see MIGRATION.md. v1 remains available at github.com/coldsmirk/go-streams and is maintained on the v1 branch.

Design

A Stream is an iterator, not a wrapper around one.

type Stream[T any] iter.Seq[T]
type Stream2[K, V any] iter.Seq2[K, V]

Because the underlying type is the standard library's, a Stream converts both ways at no cost and can be ranged over directly:

for v := range s { ... }                     // no adapter needed
sorted := slices.Sorted(iter.Seq[int](s))    // out to the standard library
s := streams.From(maps.Keys(m))              // in from it

Methods and functions split where the compiler makes them split.

A method is used when the operation needs nothing from the element type beyond any, and the result type is either unchanged or determined by a function the caller supplies. Everything else is a package function — for the same three reasons slices.Sort and slices.Chunk are functions:

Reason Examples
Needs a constraint on the element type Sort, Max, Distinct, Sum
Regroups the sequence Chunk, Window
Destructures the element type Flatten
Combines several streams Concat, Merge, Interleave

As in the standard library, each constrained function is paired with an unconstrained Func method:

top, ok := streams.Max(numbers)                  // T is cmp.Ordered
oldest, ok := people.MaxFunc(func(a, b Person) int {
    return cmp.Compare(a.Age, b.Age)             // T is anything
})

No Optional, no Result, no Pair.

Optional results use the comma-ok form Go already uses everywhere else. Fallible work uses the standard iter.Seq2[T, error] — slices, maps and iter have no error abstraction, and neither does this package. Pairs are Stream2, and three or more values travel through a combiner function:

v, ok := s.First()
lines, err := streams.Try(streams.TryMap(s, parse))
pairs := names.Zip(ages)                              // Stream2[string, int]
sums  := a.ZipWith(b, func(x, y int) int { return x + y })

Core API

Full documentation and runnable examples: pkg.go.dev, or go doc -all github.com/coldsmirk/go-streams/v2.

Constructing — Of, From, From2, Pairs, Chan, ChanContext, Range, Repeat, Iterate, Generate, Empty, Empty2

Transforming — Filter, Map, FlatMap, Scan, DistinctBy, Take, Drop, TakeWhile, DropWhile, SortFunc, SortStableFunc, CompactFunc, Reverse, Peek, KeyBy, Zip, ZipWith, Enumerate

Consuming — Collect, ForEach, Count, Fold, Reduce, First, Last, Find, Any, All, MinFunc, MaxFunc, GroupBy, IndexBy, ToMap, Partition

Package functions — Sort, Min, Max, Compact, Distinct, Contains, Frequency, Sum, Product, Average, Chunk, Window, Flatten, Concat, Merge, Interleave, Cycle, TryMap, Ok, Try

Stream2 — Keys, Values, Filter, MapKeys, MapValues, Collapse, Swap, Take, ForEach, Count, Fold

Parallel — ParallelMap, ParallelFilter, ParallelForEach, configured with WithConcurrency and Unordered. Results keep their input order unless Unordered is given.

squares := streams.Range(1, 1000).
    ParallelMap(expensive, streams.WithConcurrency(8)).
    Collect()

Subpackages

The core package imports nothing outside the standard library. Everything with a heavier dependency lives in its own package. Fallible sources return the standard iter.Seq2[T, error]; streams.Try collects one, stopping at the first error.

.../v2/source — readers, files, CSV
// streams the file; use streams.Try instead only when you want it all in a slice
lines, readErr := streams.Ok(source.LinesFile("access.log"))
n := lines.Filter(isError).Count()
if err := readErr(); err != nil {
    return err
}

for rec, err := range source.Records(r) {   // first row is the header
    if err != nil { return err }
    use(rec["email"])
}

err := source.WriteCSVFile("out.csv", rows)

Lines, LinesFile, StringLines, Bytes, Runes, CSV, CSVFile, TSV, TSVFile, Delimited, Records, RecordsFile, Keyed, File, WriteLines, WriteFile, WriteCSV, WriteCSVFile, and the Record type.

File sources open and close the file themselves — unlike v1, there is nothing for the caller to Close. An error ends the sequence.

The three concerns compose rather than multiply: Delimited picks the delimiter, Keyed applies a header row, and File handles opening. A tab-separated file with a header is Keyed(TSVFile(path)), and a format this package does not name is File(path, parse).

.../v2/temporal — time-based operators
recent := temporal.Throttle(ctx, events, 100*time.Millisecond)
batches := temporal.Tumbling(ctx, events, time.Second)

Throttle, Debounce, Sample, Delay, RateLimit, Tumbling, Sliding, Session, Timeout, Interval, Stamp.

Every operator that waits on the clock takes ctx first and stops promptly when it is done; Stamp never waits, so it takes none. Stamp returns Stream2[time.Time, T], so there is no timestamp wrapper type.

These operators read the source on a goroutine of their own, and Go cannot interrupt a goroutine parked inside a caller-supplied iterator. So feed them a source that ends when your context does, and shutdown is bounded:

events := streams.ChanContext(ctx, ch)        // not streams.Chan(ch)
recent := temporal.Throttle(ctx, events, 100*time.Millisecond)

With streams.Chan over a channel that goes quiet and is never closed, the reader stays parked until the channel produces or closes. Timers are always released either way. The package documentation states this precisely.

.../v2/join — relational joins
rows := join.Inner(orders, customers, func(id OrderID, o Order, c Customer) Row {
    return Row{Order: o, Customer: c}
})

// two unkeyed streams: derive the keys first, and every join accepts them
rows = join.Left(orders.KeyBy(byCustomer), customers.KeyBy(byID), combine)

Inner, Left, Right, Full, Group, Semi, Anti.

Every join takes a combiner and returns what it produces, so there are no JoinResult types. Outer joins pass a presence flag alongside the possibly-zero value. Each doc comment states which side is buffered.

.../v2/collections — go-collections bridge
set := collections.ToHashSet(streams.Of(ids...))   // coll.Set[int]
back := collections.FromSet(set).Collect()

sorted := collections.ToTreeSet(streams.Of(ids...), cmp.Compare)
top, _ := collections.FromSortedSet(sorted).First()

FromSet, FromSortedSet, FromList, FromQueue, FromStack, FromDeque, FromPriorityQueue, FromMap, FromSortedMap, and ToHashSet, ToTreeSet, ToArrayList, ToLinkedList, ToHashMap, ToTreeMap.

Every From* iterates the live collection lazily rather than copying it.

Semantics worth knowing

Errors live at the edges. The core has no error type, matching slices, maps and iter. A fallible source returns the standard iter.Seq2[T, error]; streams.Ok turns one into a Stream plus an error accessor, in the shape of bufio.Scanner, and streams.Try collects one into a slice when buffering is fine.

Streams are single-pass. A terminal operation consumes the underlying iterator. Traverse a Stream once; build a new one to traverse again.

Laziness is real, and short-circuiting works. First, Any, All, Find and Take stop the source as soon as they can:

// evaluates the mapping three times, not a thousand
v, _ := streams.Range(0, 1000).Map(expensive).Filter(pred).First()

The zero Stream is not valid. Use Empty[T](). A nil iterator panics when ranged over, exactly as a nil iter.Seq does.

Some operations must buffer. Sort, SortFunc, SortStableFunc, Reverse and Cycle read the whole sequence, so they cannot be used on an infinite stream. Everything else streams. Each doc comment says which.

Early termination is honoured everywhere. The iter contract panics if yield is called after it returns false; every operation in this package stops instead, and every one of them has a test that proves it.

Requirements

Go 1.27 or later. Generic methods are load-bearing; the package will not build on an earlier toolchain.

Contributing

Issues and pull requests are welcome. task check runs the formatter, go vet, the linter and the tests — the same set CI runs, plus the modernize analyzer.

Two conventions this package holds to:

  • Every operation gets an early-termination test. The iter contract panics if yield is called after it returns false, so stopping correctly is a correctness requirement, not a nicety.
  • Examples are runnable. They belong in example_test.go as Example functions with an // Output: block, not in this file, so that the compiler and CI check them.

Acknowledgments

The shape of the API follows the standard library's slices, maps and iter packages. The operator vocabulary owes to Java's Stream and Rust's Iterator.

License

MIT. See LICENSE.

Documentation

Overview

Package streams provides lazy, composable sequences.

A Stream is defined as iter.Seq, so it converts to and from a standard library iterator at no cost and may be ranged over directly: 

for v := range s { ... }

Operations split along the same line the standard library draws in slices. A method is used when the operation needs nothing from the element type beyond any, and the result type is either unchanged or determined by a function the caller supplies: 

names := streams.Of(users...).
	Filter(func(u User) bool { return u.Active }).
	Map(func(u User) string { return u.Name }).
	Collect()

A package-level function is used when the operation constrains the element type, regroups the sequence, or destructures the element type. As in slices, a constrained function is paired with an unconstrained Func method: Max requires cmp.Ordered, while Stream.MaxFunc takes a comparison function and works for any element type.

The zero Stream is not valid; use Empty for an empty sequence.

Streams are lazy and single-pass. Each terminal operation consumes the underlying iterator, so a Stream should be traversed once. Construct a new Stream to iterate again.

This package has no error abstraction, matching slices, maps and iter. Fallible sequences use the standard iter.Seq2[T, error]; see TryMap and Try for the bridge.

Example

A pipeline stays a single chain even where the element type changes. 

package main

import (
	"fmt"
	"strings"

	streams "github.com/coldsmirk/go-streams/v2"
)

type user struct {
	Name string
	Age  int
	City string
}

var users = []user{
	{"Ada", 36, "London"},
	{"Linus", 54, "Portland"},
	{"Rob", 60, "NYC"},
	{"Ken", 81, "NYC"},
}

func main() {
	lengths := streams.Of(users...).
		Filter(func(u user) bool { return u.Age > 40 }).
		Map(func(u user) string { return u.Name }).
		Map(strings.ToUpper).
		Map(func(s string) int { return len(s) }).
		Collect()

	fmt.Println(lengths)
}

Output:
[5 3 3]
Example (ErrorHandling)

Fallible work travels as iter.Seq2[T, error]. Try collects eagerly and returns the first error; Ok keeps the pipeline lazy and reports the failure once it has drained. 

package main

import (
	"fmt"
	"strconv"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	nums, err := streams.Try(streams.TryMap(streams.Of("1", "2", "3"), strconv.Atoi))
	fmt.Println(nums, err)

	_, err = streams.Try(streams.TryMap(streams.Of("1", "x", "3"), strconv.Atoi))
	fmt.Println(err)

	parsed, failed := streams.Ok(streams.TryMap(streams.Of("4", "5", "boom"), strconv.Atoi))
	fmt.Println(streams.Sum(parsed), failed() != nil)
}

Output:
[1 2 3] <nil>
strconv.Atoi: parsing "x": invalid syntax
9 true
Example (GroupAndAggregate)

GroupBy returns a plain map, which Pairs feeds back into a pipeline when the per-group work is itself a pipeline. 

byCity := streams.Of(users...).GroupBy(func(u user) string { return u.City })

for _, city := range slices.Sorted(maps.Keys(byCity)) {
	ages := streams.Of(byCity[city]...).Map(func(u user) int { return u.Age })
	avg, _ := streams.Average(ages)
	fmt.Printf("%-8s n=%d avg=%.1f\n", city, len(byCity[city]), avg)
}

Output:
London   n=1 avg=36.0
NYC      n=2 avg=70.5
Portland n=1 avg=54.0
Example (InfiniteStream)

Infinite sources are bounded by the pipeline, not by the source. Nothing past the bound is ever produced. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	powers := streams.Iterate(1, func(n int) int { return n * 2 }).
		TakeWhile(func(n int) bool { return n < 100 }).
		Collect()
	fmt.Println(powers)

	a, b := 0, 1
	fib := streams.Generate(func() int {
		a, b = b, a+b
		return a
	}).Take(8).Collect()
	fmt.Println(fib)
}

Output:
[1 2 4 8 16 32 64]
[1 1 2 3 5 8 13 21]
Example (Statistics)

A Stream is single-pass, so each statistic needs one of its own. Where several are wanted, build them from a constructor rather than reusing a Stream. 

ages := func() streams.Stream[int] {
	return streams.Of(users...).Map(func(u user) int { return u.Age })
}

lo, _ := streams.Min(ages())
hi, _ := streams.Max(ages())
avg, _ := streams.Average(ages())

fmt.Printf("n=%d sum=%d range=%d..%d avg=%.2f\n",
	ages().Count(), streams.Sum(ages()), lo, hi, avg)

Output:
n=4 sum=231 range=36..81 avg=57.75
Example (StdlibInterop)

A Stream is an iter.Seq, so it crosses into and out of the standard library with a conversion rather than an adapter. 

package main

import (
	"cmp"
	"fmt"
	"iter"
	"maps"
	"slices"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	s := streams.Of(3, 1, 2)
	fmt.Println(slices.Sorted(iter.Seq[int](s)))

	m := map[string]int{"b": 2, "a": 1}
	fmt.Println(streams.From(maps.Keys(m)).SortFunc(cmp.Compare).Collect())

	for i, name := range streams.Of("x", "y").Enumerate() {
		fmt.Printf("%d:%s ", i, name)
	}
	fmt.Println()
}

Output:
[1 2 3]
[a b]
0:x 1:y
Example (WordFrequency)

Counting and ranking. Frequency builds the histogram, Pairs puts the map back into a pipeline, and SortFunc with Take ranks it. Comparing the word after the count keeps ties in a defined order. 

package main

import (
	"cmp"
	"fmt"
	"strings"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	const text = "the quick brown fox jumps over the lazy dog the fox"

	type count struct {
		word string
		n    int
	}

	top := streams.Pairs(streams.Frequency(streams.Of(strings.Fields(text)...))).
		Collapse(func(w string, n int) count { return count{w, n} }).
		SortFunc(func(a, b count) int {
			if c := cmp.Compare(b.n, a.n); c != 0 {
				return c
			}
			return cmp.Compare(a.word, b.word)
		}).
		Take(3).
		Collect()

	for _, c := range top {
		fmt.Println(c.word, c.n)
	}
}

Output:
the 3
fox 2
brown 1

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func Average 

func Average[T Numeric](s Stream[T]) (float64, bool)

Average returns the arithmetic mean of the elements, or false if the Stream is empty.

func Contains 

func Contains[T comparable](s Stream[T], target T) bool

Contains reports whether any element equals target. It stops at the first match.

func Frequency 

func Frequency[T comparable](s Stream[T]) map[T]int

Frequency returns a map from each distinct element to the number of times it appears.

func Max 

func Max[T cmp.Ordered](s Stream[T]) (T, bool)

Max returns the largest element, or false if the Stream is empty.

func Min 

func Min[T cmp.Ordered](s Stream[T]) (T, bool)

Min returns the smallest element, or false if the Stream is empty.

func Product 

func Product[T Numeric](s Stream[T]) T

Product returns the product of the elements, or one for an empty Stream.

func Sum 

func Sum[T Numeric](s Stream[T]) T

Sum returns the sum of the elements, or the zero value for an empty Stream.

func Try 

func Try[T any](seq iter.Seq2[T, error]) ([]T, error)

Try collects the values of seq, stopping at and returning the first error.

func TryMap 

func TryMap[T, R any](s Stream[T], fn func(T) (R, error)) iter.Seq2[R, error]

TryMap applies fn to each element and returns the results paired with the error fn reported. The result is a plain iter.Seq2, so it can be ranged over directly: 

for v, err := range streams.TryMap(s, parse) { ... }
Example

The package has no error type. Fallible work uses the standard iter.Seq2[T, error]. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	parse := func(s string) (int, error) {
		if s == "" {
			return 0, fmt.Errorf("empty field")
		}
		return len(s), nil
	}

	got, err := streams.Try(streams.TryMap(streams.Of("ab", "cde"), parse))
	fmt.Println(got, err)

	got, err = streams.Try(streams.TryMap(streams.Of("ab", "", "cde"), parse))
	fmt.Println(got, err)
}

Output:
[2 3] <nil>
[2] empty field

Types

type Numeric 

type Numeric interface {
	~int | ~int8 | ~int16 | ~int32 | ~int64 |
		~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 | ~uintptr |
		~float32 | ~float64
}

Numeric is a constraint for the types that support the arithmetic used by Sum, Product and Average. Complex types are excluded: they are not ordered and an average over them is rarely meaningful.

type ParallelOption 

type ParallelOption func(*parallelConfig)

A ParallelOption configures a parallel operation.

func Unordered 

func Unordered() ParallelOption

Unordered lets results be emitted as they finish rather than in the order of the input. It is faster when the work per element varies.

func WithConcurrency 

func WithConcurrency(n int) ParallelOption

WithConcurrency sets how many elements may be processed at once. The default is runtime.GOMAXPROCS(0). Values below one are treated as one.

type Stream 

type Stream[T any] iter.Seq[T]

Stream is a lazy sequence of values. Its underlying type is iter.Seq[T], so a Stream may be converted to and from iter.Seq at no cost and ranged over directly.

Example

A Stream is an iter.Seq, so it may be ranged over directly. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	for v := range streams.Range(0, 3) {
		fmt.Print(v, " ")
	}
	fmt.Println()
}

Output:
0 1 2

func Chan 

func Chan[T any](ch <-chan T) Stream[T]

Chan returns a Stream over the values received from ch, ending when ch is closed. Stopping the iteration early leaves any unreceived values in ch.

A Stream is pulled, so the only way out of a receive that never completes is for ch to produce or close. An operator that reads the Stream on a goroutine of its own, as those in github.com/coldsmirk/go-streams/v2/temporal do, therefore cannot release that goroutine while ch stays quiet. Use ChanContext for a pipeline that must shut down on cancellation alone.

func ChanContext 

func ChanContext[T any](ctx context.Context, ch <-chan T) Stream[T]

ChanContext returns a Stream over the values received from ch, ending when ch is closed or ctx is done, whichever happens first. Stopping the iteration early leaves any unreceived values in ch.

Prefer it to Chan for a long-lived source. Because the receive and the cancellation are one select, a goroutine parked in this Stream is always released by cancelling ctx — including the goroutine a temporal operator uses to read its source, which Chan leaves parked until ch produces or closes. Pass the same context to the source and to the operator reading it.

func Chunk 

func Chunk[T any](s Stream[T], n int) Stream[[]T]

Chunk returns a Stream of consecutive, non-overlapping slices of up to n elements. The final slice is short if the Stream does not divide evenly. Chunk panics if n is not positive.

Example

Chunk regroups the sequence, so it is a package function, mirroring slices.Chunk. The chain resumes off its result. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	sizes := streams.Chunk(streams.Range(0, 5), 2).
		Map(func(c []int) int { return len(c) }).
		Collect()
	fmt.Println(sizes)
}

Output:
[2 2 1]

func Compact 

func Compact[T comparable](s Stream[T]) Stream[T]

Compact returns a Stream omitting each element equal to the one before it. Like slices.Compact, it removes only adjacent duplicates.

func Concat 

func Concat[T any](ss ...Stream[T]) Stream[T]

Concat returns a Stream of the elements of each Stream in turn.

func Cycle 

func Cycle[T any](s Stream[T]) Stream[T]

Cycle returns an infinite Stream repeating the elements of s. It buffers the elements on the first pass, so s must be finite.

func Distinct 

func Distinct[T comparable](s Stream[T]) Stream[T]

Distinct returns a Stream omitting every element that has already appeared. Elements are held for the duration of the iteration.

func Empty 

func Empty[T any]() Stream[T]

Empty returns a Stream with no elements.

func Flatten 

func Flatten[T any](s Stream[Stream[T]]) Stream[T]

Flatten returns a Stream of the elements of every Stream in s. It destructures the element type, so it cannot be a method.

func From 

func From[T any](seq iter.Seq[T]) Stream[T]

From returns seq as a Stream. It is a conversion that infers T, which is its only purpose: Stream[T](seq) requires naming T, From(seq) does not.

Example

Streams convert to and from standard library iterators at no cost. 

package main

import (
	"fmt"
	"iter"
	"maps"
	"slices"
	"strings"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	m := map[string]int{"b": 2, "a": 1, "c": 3}

	// in: maps.Keys returns an iter.Seq, From infers the element type
	names := streams.From(maps.Keys(m)).
		Filter(func(s string) bool { return s != "b" }).
		SortFunc(strings.Compare).
		Collect()
	fmt.Println(names)

	// out: a Stream converts straight back to an iter.Seq
	fmt.Println(slices.Sorted(iter.Seq[int](streams.Pairs(m).Values())))
}

Output:
[a c]
[1 2 3]

func Generate 

func Generate[T any](fn func() T) Stream[T]

Generate returns an infinite Stream of values produced by fn.

func Interleave 

func Interleave[T any](a, b Stream[T]) Stream[T]

Interleave returns a Stream alternating between the elements of a and b. When one is exhausted the remainder of the other follows.

func Iterate 

func Iterate[T any](seed T, next func(T) T) Stream[T]

Iterate returns an infinite Stream of seed, next(seed), next(next(seed)), ...

func Merge 

func Merge[T any](compare func(a, b T) int, ss ...Stream[T]) Stream[T]

Merge returns a Stream of the elements of every Stream in ss in the order given by compare. Each input must already be sorted by compare; if it is not, the output is not sorted either.

func Of 

func Of[T any](values ...T) Stream[T]

Of returns a Stream over values. Pass a slice with s... to stream it.

func Ok 

func Ok[T any](seq iter.Seq2[T, error]) (Stream[T], func() error)

Ok returns a Stream of the values of seq up to its first error, together with a function reporting that error. Consume the Stream, then call the function, the way bufio.Scanner pairs Scan with Err: 

lines, readErr := streams.Ok(source.LinesFile(path))
n := lines.Filter(nonBlank).Count()
if err := readErr(); err != nil {
	return err
}

It is the lazy counterpart of Try. Try holds every value in a slice, so it cannot read a source larger than memory; Ok lets a fallible source feed a pipeline one element at a time. The Stream ends at the first error, and the function reports nil until one is reached, so it is only meaningful once the Stream has been consumed. Like every Stream it is single-pass, and like bufio.Scanner it is not safe to consume from one goroutine while calling the function from another.

func Range 

func Range(start, end int) Stream[int]

Range returns a Stream over the integers in [start, end).

func Repeat 

func Repeat[T any](value T, n int) Stream[T]

Repeat returns a Stream containing value n times. If n is negative, the Stream is infinite.

func Sort 

func Sort[T cmp.Ordered](s Stream[T]) Stream[T]

Sort returns a Stream of the elements in ascending order. It buffers the whole sequence, so it is not usable on an infinite Stream. For an element type that is not ordered, use Stream.SortFunc.

func Window 

func Window[T any](s Stream[T], n int) Stream[[]T]

Window returns a Stream of overlapping slices of n consecutive elements, advancing one element at a time. It yields nothing if the Stream is shorter than n. Window panics if n is not positive.

The windows are cut from a shared backing array rather than allocated one at a time, which is what keeps a sliding window from costing an allocation and a copy per element. Two windows therefore share the elements they overlap on: reading them is unaffected, but writing through one window is visible in its neighbours, and holding on to one keeps a block of elements alive. No window is ever overwritten once yielded. Where independent slices are wanted, ask for them: 

streams.Window(s, 16).Map(slices.Clone)

Chunk does allocate per chunk, because its chunks do not overlap and sharing an array measurably costs more there than it saves.

func (Stream[T]) All 

func (s Stream[T]) All(pred func(T) bool) bool

All reports whether pred is true for every element. It stops at the first element that fails, and is true for an empty Stream.

func (Stream[T]) Any 

func (s Stream[T]) Any(pred func(T) bool) bool

Any reports whether pred is true for at least one element. It stops at the first match.

func (Stream[T]) Collect 

func (s Stream[T]) Collect() []T

Collect returns a slice of the elements.

func (Stream[T]) CompactFunc 

func (s Stream[T]) CompactFunc(eq func(a, b T) bool) Stream[T]

CompactFunc returns a Stream omitting each element that eq reports equal to the element before it. Like slices.CompactFunc, it removes only adjacent duplicates; use Distinct or Stream.DistinctBy to remove all of them.

func (Stream[T]) Count 

func (s Stream[T]) Count() int

Count returns the number of elements.

func (Stream[T]) DistinctBy 

func (s Stream[T]) DistinctBy[K comparable](key func(T) K) Stream[T]

DistinctBy returns a Stream omitting elements whose key has already been seen. Keys are held for the duration of the iteration.

func (Stream[T]) Drop 

func (s Stream[T]) Drop(n int) Stream[T]

Drop returns a Stream omitting the first n elements.

func (Stream[T]) DropWhile 

func (s Stream[T]) DropWhile(pred func(T) bool) Stream[T]

DropWhile returns a Stream omitting the leading elements for which pred reports true.

func (Stream[T]) Enumerate 

func (s Stream[T]) Enumerate() Stream2[int, T]

Enumerate returns a Stream2 pairing each element with its zero-based index.

func (Stream[T]) Filter 

func (s Stream[T]) Filter(keep func(T) bool) Stream[T]

Filter returns a Stream of the elements for which keep reports true.

func (Stream[T]) Find 

func (s Stream[T]) Find(pred func(T) bool) (T, bool)

Find returns the first element for which pred reports true.

func (Stream[T]) First 

func (s Stream[T]) First() (T, bool)

First returns the first element, or false if the Stream is empty.

func (Stream[T]) FlatMap 

func (s Stream[T]) FlatMap[R any](fn func(T) Stream[R]) Stream[R]

FlatMap returns a Stream of the elements of every Stream produced by fn.

func (Stream[T]) Fold 

func (s Stream[T]) Fold[A any](init A, fn func(A, T) A) A

Fold accumulates the elements into a single value of any type, starting from init. Unlike Stream.Reduce, the accumulator need not be the element type.

Example

Fold accumulates into a type unrelated to the element type. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

type user struct {
	Name string
	Age  int
	City string
}

var users = []user{
	{"Ada", 36, "London"},
	{"Linus", 54, "Portland"},
	{"Rob", 60, "NYC"},
	{"Ken", 81, "NYC"},
}

func main() {
	total := streams.Of(users...).Fold(0, func(sum int, u user) int {
		return sum + u.Age
	})
	fmt.Println(total)
}

Output:
231

func (Stream[T]) ForEach 

func (s Stream[T]) ForEach(fn func(T))

ForEach calls fn for each element.

func (Stream[T]) GroupBy 

func (s Stream[T]) GroupBy[K comparable](key func(T) K) map[K][]T

GroupBy returns a map from each key to the elements that produced it, in encounter order.

Example 
package main

import (
	"fmt"
	"maps"
	"slices"

	streams "github.com/coldsmirk/go-streams/v2"
)

type user struct {
	Name string
	Age  int
	City string
}

var users = []user{
	{"Ada", 36, "London"},
	{"Linus", 54, "Portland"},
	{"Rob", 60, "NYC"},
	{"Ken", 81, "NYC"},
}

func main() {
	byCity := streams.Of(users...).GroupBy(func(u user) string { return u.City })
	for _, city := range slices.Sorted(maps.Keys(byCity)) {
		fmt.Printf("%s:%d ", city, len(byCity[city]))
	}
	fmt.Println()
}

Output:
London:1 NYC:2 Portland:1

func (Stream[T]) IndexBy 

func (s Stream[T]) IndexBy[K comparable](key func(T) K) map[K]T

IndexBy returns a map from each key to the last element that produced it.

func (Stream[T]) KeyBy 

func (s Stream[T]) KeyBy[K any](key func(T) K) Stream2[K, T]

KeyBy returns a Stream2 pairing each element with the key that key derives from it. It is the general way from a Stream to a Stream2, as Stream2.Collapse is the general way back; Stream.Enumerate and Stream.Zip are the two pairings a key function cannot express.

func (Stream[T]) Last 

func (s Stream[T]) Last() (T, bool)

Last returns the last element, or false if the Stream is empty.

func (Stream[T]) Map 

func (s Stream[T]) Map[R any](fn func(T) R) Stream[R]

Map returns a Stream of the results of applying fn to each element.

Example

Map is a generic method, so the result element type follows the function. 

package main

import (
	"fmt"
	"strings"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	names := streams.Of(1, 2, 3).Map(func(i int) string {
		return strings.Repeat("*", i)
	}).Collect()
	fmt.Println(names)
}

Output:
[* ** ***]

func (Stream[T]) MaxFunc 

func (s Stream[T]) MaxFunc(compare func(a, b T) int) (T, bool)

MaxFunc returns the maximal element as ordered by compare, or false if the Stream is empty. If several elements are maximal, it returns the first.

Example

MaxFunc works for any element type; Max is its constrained counterpart, in the same way slices.MaxFunc pairs with slices.Max. 

package main

import (
	"cmp"
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

type user struct {
	Name string
	Age  int
	City string
}

var users = []user{
	{"Ada", 36, "London"},
	{"Linus", 54, "Portland"},
	{"Rob", 60, "NYC"},
	{"Ken", 81, "NYC"},
}

func main() {
	oldest, ok := streams.Of(users...).MaxFunc(func(a, b user) int {
		return cmp.Compare(a.Age, b.Age)
	})
	fmt.Println(oldest.Name, ok)

	largest, ok := streams.Max(streams.Of(3, 9, 4))
	fmt.Println(largest, ok)
}

Output:
Ken true
9 true

func (Stream[T]) MinFunc 

func (s Stream[T]) MinFunc(compare func(a, b T) int) (T, bool)

MinFunc returns the minimal element as ordered by compare, or false if the Stream is empty. If several elements are minimal, it returns the first.

func (Stream[T]) ParallelFilter 

func (s Stream[T]) ParallelFilter(pred func(T) bool, opts ...ParallelOption) Stream[T]

ParallelFilter is Stream.Filter with pred evaluated concurrently. Elements keep their input order unless Unordered is given. pred must be safe for concurrent use. The note on Stream.ParallelMap about when concurrency pays applies here too.

func (Stream[T]) ParallelForEach 

func (s Stream[T]) ParallelForEach(fn func(T), opts ...ParallelOption)

ParallelForEach calls fn for each element concurrently and returns once every call has completed. fn must be safe for concurrent use. Unordered has no effect here, since there are no results to order.

func (Stream[T]) ParallelMap 

func (s Stream[T]) ParallelMap[R any](fn func(T) R, opts ...ParallelOption) Stream[R]

ParallelMap is Stream.Map with fn applied concurrently. Results keep their input order unless Unordered is given. fn must be safe for concurrent use.

Concurrency is not free: each element costs a goroutine and a channel handoff, which is on the order of a microsecond. Below roughly five microseconds of work per element this is slower than Stream.Map, and for trivial work it is slower by orders of magnitude. Measure before reaching for it.

Example

ParallelMap keeps the input order unless Unordered is given. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	squares := streams.Range(1, 6).
		ParallelMap(func(i int) int { return i * i }, streams.WithConcurrency(4)).
		Collect()
	fmt.Println(squares)
}

Output:
[1 4 9 16 25]

func (Stream[T]) Partition 

func (s Stream[T]) Partition(pred func(T) bool) (yes, no []T)

Partition returns the elements for which pred reports true and false.

func (Stream[T]) Peek 

func (s Stream[T]) Peek(fn func(T)) Stream[T]

Peek returns a Stream that calls fn for each element as it passes through. It is intended for tracing a pipeline, not for mutating it.

func (Stream[T]) Reduce 

func (s Stream[T]) Reduce(fn func(a, b T) T) (T, bool)

Reduce combines the elements using fn, returning false if the Stream is empty.

func (Stream[T]) Reverse 

func (s Stream[T]) Reverse() Stream[T]

Reverse returns a Stream of the elements in reverse order. It buffers the whole sequence, so it is not usable on an infinite Stream.

func (Stream[T]) Scan 

func (s Stream[T]) Scan[A any](init A, fn func(A, T) A) Stream[A]

Scan returns a Stream of the successive accumulated values, starting with the result of combining init with the first element. It is Fold that yields every intermediate accumulator rather than only the last.

func (Stream[T]) SortFunc 

func (s Stream[T]) SortFunc(compare func(a, b T) int) Stream[T]

SortFunc returns a Stream of the elements sorted by compare. It buffers the whole sequence, so it is not usable on an infinite Stream.

func (Stream[T]) SortStableFunc 

func (s Stream[T]) SortStableFunc(compare func(a, b T) int) Stream[T]

SortStableFunc is like Stream.SortFunc but keeps equal elements in their original order.

func (Stream[T]) Take 

func (s Stream[T]) Take(n int) Stream[T]

Take returns a Stream of at most the first n elements.

func (Stream[T]) TakeWhile 

func (s Stream[T]) TakeWhile(pred func(T) bool) Stream[T]

TakeWhile returns a Stream of the leading elements for which pred reports true, stopping at the first element for which it reports false.

func (Stream[T]) ToMap 

func (s Stream[T]) ToMap[K comparable, V any](fn func(T) (K, V)) map[K]V

ToMap returns a map built from the key-value pairs fn derives from each element. Later pairs overwrite earlier ones with the same key.

func (Stream[T]) Zip 

func (s Stream[T]) Zip[U any](o Stream[U]) Stream2[T, U]

Zip returns a Stream2 pairing each element of s with the element of o at the same position, ending when either is exhausted.

Example

Zip yields a Stream2, so the package needs no pair type. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	paired := streams.Of("a", "b").Zip(streams.Of(1, 2))
	for k, v := range paired {
		fmt.Printf("%s=%d ", k, v)
	}
	fmt.Println()
}

Output:
a=1 b=2

func (Stream[T]) ZipWith 

func (s Stream[T]) ZipWith[U, R any](o Stream[U], fn func(T, U) R) Stream[R]

ZipWith returns a Stream of fn applied to the elements of s and o at the same position, ending when either is exhausted. It is Stream.Zip followed by Stream2.Collapse; neither materialises a pair, since a Stream2 passes its two values to yield separately.

Example

ZipWith fuses the pairing with the mapping and allocates nothing in between. 

package main

import (
	"fmt"

	streams "github.com/coldsmirk/go-streams/v2"
)

func main() {
	sums := streams.Of(1, 2, 3).ZipWith(streams.Of(10, 20, 30),
		func(a, b int) int { return a + b }).Collect()
	fmt.Println(sums)
}

Output:
[11 22 33]

type Stream2 

type Stream2[K, V any] iter.Seq2[K, V]

Stream2 is a lazy sequence of paired values, conventionally key-value. Its underlying type is iter.Seq2[K, V].

func Empty2 

func Empty2[K, V any]() Stream2[K, V]

Empty2 returns a Stream2 with no elements.

func From2 

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

From2 returns seq as a Stream2. Like From, it exists to infer K and V.

func Pairs 

func Pairs[M ~map[K]V, K comparable, V any](m M) Stream2[K, V]

Pairs returns a Stream2 over the key-value pairs of m, in unspecified order.

func (Stream2[K, V]) Collapse 

func (s Stream2[K, V]) Collapse[R any](fn func(K, V) R) Stream[R]

Collapse returns a Stream of fn applied to each pair. It is the way back from a Stream2 to a Stream.

func (Stream2[K, V]) Count 

func (s Stream2[K, V]) Count() int

Count returns the number of pairs.

func (Stream2[K, V]) Filter 

func (s Stream2[K, V]) Filter(keep func(K, V) bool) Stream2[K, V]

Filter returns a Stream2 of the pairs for which keep reports true.

func (Stream2[K, V]) Fold 

func (s Stream2[K, V]) Fold[A any](init A, fn func(A, K, V) A) A

Fold accumulates the pairs into a single value, starting from init.

func (Stream2[K, V]) ForEach 

func (s Stream2[K, V]) ForEach(fn func(K, V))

ForEach calls fn for each pair.

func (Stream2[K, V]) Keys 

func (s Stream2[K, V]) Keys() Stream[K]

Keys returns a Stream of the first element of each pair.

func (Stream2[K, V]) MapKeys 

func (s Stream2[K, V]) MapKeys[J any](fn func(K) J) Stream2[J, V]

MapKeys returns a Stream2 with fn applied to each key.

func (Stream2[K, V]) MapValues 

func (s Stream2[K, V]) MapValues[W any](fn func(V) W) Stream2[K, W]

MapValues returns a Stream2 with fn applied to each value.

func (Stream2[K, V]) Swap 

func (s Stream2[K, V]) Swap() Stream2[V, K]

Swap returns a Stream2 with the elements of each pair exchanged.

func (Stream2[K, V]) Take 

func (s Stream2[K, V]) Take(n int) Stream2[K, V]

Take returns a Stream2 of at most the first n pairs.

func (Stream2[K, V]) Values 

func (s Stream2[K, V]) Values() Stream[V]

Values returns a Stream of the second element of each pair.

Source Files

View all Source files

Directories

Path Synopsis
Package collections bridges streams and the container types of github.com/coldsmirk/go-collections.
 Package collections bridges streams and the container types of github.com/coldsmirk/go-collections.
Package join provides relational joins over keyed streams.
 Package join provides relational joins over keyed streams.
Package source reads sequences from readers and files, and writes them back out.
 Package source reads sequences from readers and files, and writes them back out.
Package temporal provides time-based stream operators.
 Package temporal provides time-based stream operators.

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.