streams

README

Streams

This package lets you use Java-like generic streams in Go. Streams can be either infinite or finite. Values are lazy-fetched, and therefore, values aren't processed until terminal methods are invoked.

It is worth to mention that there are two types of streams:

• Stream[T] - for single-value streams
• Stream2[T, U] - for streams that return two values at once

Construction

You can create a stream in multiple ways:

1. Manually (by calling From or From2 method with a seq function as argument)
2. Streaming object (by calling Stream or Stream2 method on an object that implements Streamer or Streamer2 interface)
3. Constructor (by calling one of the provided constructor functions):
   • FromChannel(ch <-chan T) Stream[T]
   • Range(lower, upper int) Stream[int]
   • FromGenerator[T any](generator func() T) Stream[T]

Intermediates

There are methods that you can use to transform single-value streams:

• Map(mapper functions.Mapper[T, U]) Stream[U]
• Filter(predicate functions.predicate[T]) Stream[T]
• Limit(count int) Stream[T]
• Seek(count int) Stream[T]
• Sort(comparator functions.Comparator[T]) Stream[T]
• Window(width int) Stream[[]T]

And methods that you can use to transform two-value streams:

• Keys() Stream[K]
• Values() Stream[V]
• FilterKeys(predicate functions.predicate[K]) Stream2[K, V]
• FilterValues(predicate functions.predicate[V]) Stream2[K, V]
• Limit(count int) Stream2[K, V]
• Seek(count int) Stream2[K, V]

Terminals

After you have transformed the stream, you can use terminal methods to process values in the stream and get the result that you wanted.

Methods that you can use to process single-value streams:

• ForEach(function functions.ParamCallback[T])
• Reduce(accumulator P, reducer functions.Reducer[T, P]) P
• Any(predicate functions.predicate[T]) bool
• All(predicate functions.predicate[T]) bool
• Collect[T, R any](collector Collector[T, R]) R
• Count() int
• Max(comparator functions.Comparator[T]) optional.Optional[T]
• Min(comparator functions.Comparator[T]) optional.Optional[T]
• First() optional.Optional[T]
• Find(predicate functions.predicate[T]) optional.Optional[T]
• Channel() <-chan T

Methods that you can use to process two-value streams:

• Count() int

Limitations

Due to lack of generic methods in Go, some methods (like Map and Reduce) currently only support returning same type like stream values are. If your transformations return some other type, then you should consider functions with the same name that accept the stream as the first argument.

//Instead of
stream.Map(mapFunc)

//You should use
streams.Map(stream, mapFunc)

Also, because of cyclic generic instantiations, some methods (like Window) are also functions that accept the stream as first argument.

Documentation

Index

• func Collect[T, R any](s Stream[T], collector Collector[T, R]) R
• func Reduce[T, P any](s Stream[T], acc P, reducer Reducer[T, P]) P
• type Collector
• type Reducer
• type Stream
  • func From[T any](seq iter.Seq[T]) Stream[T]
  • func FromChannel[T any](ch <-chan T) Stream[T]
  • func FromGenerator[T any](generator func() T) Stream[T]
  • func FromReader(reader io.Reader, splitFunc bufio.SplitFunc) Stream[string]
  • func Map[T, U any](s Stream[T], mapper functions.Mapper[T, U]) Stream[U]
  • func Range[T math.Real](lower, upper T) Stream[T]
  • func RangeIncrement[T math.Real](lower, upper, increment T) Stream[T]
  • func Window[T any](s Stream[T], width int) Stream[[]T]
  • func (s Stream[T]) All(predicate predication.Predicate[T]) bool
  • func (s Stream[T]) Any(predicate predication.Predicate[T]) bool
  • func (s Stream[T]) Channel() <-chan T
  • func (s Stream[T]) Collect(c Collector[T, T]) T
  • func (s Stream[T]) Count() int
  • func (s Stream[T]) Enumerate() Stream2[int, T]
  • func (s Stream[T]) Filter(predicate predication.Predicate[T]) Stream[T]
  • func (s Stream[T]) Find(predicate predication.Predicate[T]) (T, bool)
  • func (s Stream[T]) First() (T, bool)
  • func (s Stream[T]) ForEach(f functions.ParamCallback[T])
  • func (s Stream[T]) Limit(count int) Stream[T]
  • func (s Stream[T]) Map(mapper functions.Mapper[T, T]) Stream[T]
  • func (s Stream[T]) Max(comparator comparison.Comparator[T]) (T, bool)
  • func (s Stream[T]) Min(comparator comparison.Comparator[T]) (T, bool)
  • func (s Stream[T]) Reduce(acc T, reducer Reducer[T, T]) T
  • func (s Stream[T]) Seek(count int) Stream[T]
  • func (s Stream[T]) Sort(comparator comparison.Comparator[T]) Stream[T]
• type Stream2
  • func From2[K, V any](seq iter.Seq2[K, V]) Stream2[K, V]
  • func (s Stream2[K, V]) Count() int
  • func (s Stream2[K, V]) FilterKeys(predicate predication.Predicate[K]) Stream2[K, V]
  • func (s Stream2[K, V]) FilterValues(predicate predication.Predicate[V]) Stream2[K, V]
  • func (s Stream2[K, V]) Keys() Stream[K]
  • func (s Stream2[K, V]) Limit(count int) Stream2[K, V]
  • func (s Stream2[K, V]) Seek(count int) Stream2[K, V]
  • func (s Stream2[K, V]) Values() Stream[V]
• type Streamer
• type Streamer2

Functions

func Collect

func Collect[T, R any](s Stream[T], collector Collector[T, R]) R

func Reduce

func Reduce[T, P any](s Stream[T], acc P, reducer Reducer[T, P]) P

Types

type Collector

type Collector[T, R any] interface {
	Supply(value T)
	Finish() R
}

type Reducer

type Reducer[T, P any] func(acc P, value T) P

type Stream

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

func From

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

func FromChannel

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

func FromGenerator

func FromGenerator[T any](generator func() T) Stream[T]

func FromReader

func FromReader(reader io.Reader, splitFunc bufio.SplitFunc) Stream[string]

func Map

func Map[T, U any](s Stream[T], mapper functions.Mapper[T, U]) Stream[U]

func Range

func Range[T math.Real](lower, upper T) Stream[T]

func RangeIncrement

func RangeIncrement[T math.Real](lower, upper, increment T) Stream[T]

func Window

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

func (Stream[T]) All

func (s Stream[T]) All(predicate predication.Predicate[T]) bool

func (Stream[T]) Any

func (s Stream[T]) Any(predicate predication.Predicate[T]) bool

func (Stream[T]) Channel

func (s Stream[T]) Channel() <-chan T

func (Stream[T]) Collect

func (s Stream[T]) Collect(c Collector[T, T]) T

func (Stream[T]) Count

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

func (Stream[T]) Enumerate

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

func (Stream[T]) Filter

func (s Stream[T]) Filter(predicate predication.Predicate[T]) Stream[T]

func (Stream[T]) Find

func (s Stream[T]) Find(predicate predication.Predicate[T]) (T, bool)

func (Stream[T]) First

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

func (Stream[T]) ForEach

func (s Stream[T]) ForEach(f functions.ParamCallback[T])

func (Stream[T]) Limit

func (s Stream[T]) Limit(count int) Stream[T]

func (Stream[T]) Map

func (s Stream[T]) Map(mapper functions.Mapper[T, T]) Stream[T]

func (Stream[T]) Max

func (s Stream[T]) Max(comparator comparison.Comparator[T]) (T, bool)

func (Stream[T]) Min

func (s Stream[T]) Min(comparator comparison.Comparator[T]) (T, bool)

func (Stream[T]) Reduce

func (s Stream[T]) Reduce(acc T, reducer Reducer[T, T]) T

func (Stream[T]) Seek

func (s Stream[T]) Seek(count int) Stream[T]

func (Stream[T]) Sort

func (s Stream[T]) Sort(comparator comparison.Comparator[T]) Stream[T]

type Stream2

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

func From2

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

func (Stream2[K, V]) Count

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

func (Stream2[K, V]) FilterKeys

func (s Stream2[K, V]) FilterKeys(predicate predication.Predicate[K]) Stream2[K, V]

func (Stream2[K, V]) FilterValues

func (s Stream2[K, V]) FilterValues(predicate predication.Predicate[V]) Stream2[K, V]

func (Stream2[K, V]) Keys

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

func (Stream2[K, V]) Limit

func (s Stream2[K, V]) Limit(count int) Stream2[K, V]

func (Stream2[K, V]) Seek

func (s Stream2[K, V]) Seek(count int) Stream2[K, V]

func (Stream2[K, V]) Values

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

type Streamer

type Streamer[T any] interface {
	Stream(yield func(T) bool)
}

type Streamer2

type Streamer2[K, V any] interface {
	Stream2(yield func(K, V) bool)
}