stream

Documentation

Index

• func Collect[V any](input Stream[V]) ([]V, error)
• func ForEach[V any](input Stream[V], fn func(V) error) error
• func Partition[V any](input Stream[V], goLeft func(V) (bool, error)) (left Stream[V], right Stream[V])
• func Reduce[V, A any](input Stream[V], reducer func(acc A, value V) (A, error)) (A, error)
• func Sum[V numeric](input Stream[V]) (V, error)
• func SumToString[V numeric](input Stream[V]) (string, error)
• type LookaheadStream
  • func Lookahead[V any](input Stream[V]) *LookaheadStream[V]
  • func (l *LookaheadStream[V]) Next() (next V, err error)
  • func (l *LookaheadStream[V]) PeekN(n int) (peek []V, err error)
• type ResettableStream
  • func Resettable[V any](input Stream[V]) ResettableStream[V]
• type Stream
  • func DebugPrint[V any](log zerolog.Logger, stageName string, stream Stream[V]) Stream[V]
  • func Filter[A any](input Stream[A], predicate func(A) (keep bool, err error)) Stream[A]
  • func FlatMap[A, B any](input Stream[A], fn func(A) (Stream[B], error)) Stream[B]
  • func From[V any](input []V) Stream[V]
  • func FromItem[V any](input V) Stream[V]
  • func LinesFrom(input []byte) Stream[string]
  • func Map[A, B any](input Stream[A], fn func(A) (B, error)) Stream[B]
  • func SplitBy(input []byte, split byte) Stream[string]

Functions

func Collect

func Collect[V any](input Stream[V]) ([]V, error)

Collect returns all values from the stream as a slice.

func ForEach

func ForEach[V any](input Stream[V], fn func(V) error) error

ForEach applies the given function to each value in the stream.

func Partition

func Partition[V any](input Stream[V], goLeft func(V) (bool, error)) (left Stream[V], right Stream[V])

Partition splits a stream into two streams based on the result of the goLeft function

func Reduce

func Reduce[V, A any](input Stream[V], reducer func(acc A, value V) (A, error)) (A, error)

Reduce returns a single value from the stream by applying the reducer function to each value in the stream.

func Sum

func Sum[V numeric](input Stream[V]) (V, error)

Sum returns the sum of all values in the stream.

func SumToString

func SumToString[V numeric](input Stream[V]) (string, error)

SumToString returns the sum of all values in the stream as a string.

Types

type LookaheadStream

type LookaheadStream[V any] struct {
	// contains filtered or unexported fields
}

func Lookahead

func Lookahead[V any](input Stream[V]) *LookaheadStream[V]

Lookahead returns a stream in which we can look ahead from the current point

func (*LookaheadStream[V]) Next

func (l *LookaheadStream[V]) Next() (next V, err error)

func (*LookaheadStream[V]) PeekN

func (l *LookaheadStream[V]) PeekN(n int) (peek []V, err error)

type ResettableStream

type ResettableStream[V any] interface {
	Stream[V]
	Save()    // Save the current position of the stream
	Restore() // Restore the stream to the last saved position (if no save has been made then this will be the beginning)
	Reset()   // Reset the stream to the beginning
}

func Resettable

func Resettable[V any](input Stream[V]) ResettableStream[V]

type Stream

type Stream[V any] interface {
	// Next returns the next value in the stream
	// or if the stream has finished returns [io.EOF]
	Next() (V, error)
}

func DebugPrint

func DebugPrint[V any](log zerolog.Logger, stageName string, stream Stream[V]) Stream[V]

func Filter

func Filter[A any](input Stream[A], predicate func(A) (keep bool, err error)) Stream[A]

Filter returns a stream with only the elements that match the given predicate.

If the predicate returns an error the stream will stop and the error will be

func FlatMap

func FlatMap[A, B any](input Stream[A], fn func(A) (Stream[B], error)) Stream[B]

FlatMap returns a new merged stream with the given function applied to each element and the results merged into a single stream.

If the function returns an error the stream will stop and the error will be returned.

func From

func From[V any](input []V) Stream[V]

From returns a stream from a given slice

func FromItem

func FromItem[V any](input V) Stream[V]

func LinesFrom

func LinesFrom(input []byte) Stream[string]

LinesFrom returns a stream of lines from the given input.

func Map

func Map[A, B any](input Stream[A], fn func(A) (B, error)) Stream[B]

Map returns a new stream with the given function applied to each element.

If the function returns an error the stream will stop and the error will be returned.

func SplitBy

func SplitBy(input []byte, split byte) Stream[string]

SplitBy returns a stream of strings split by the given byte.