streaming

Documentation

Index

• Variables
• func Collect[I any](ctx context.Context, stream Stream[I]) (slice []I, err error)
• type Mapper
• type Selector
• type Sizer
• type Stream
  • func Delay[I any](source Stream[I], delay time.Duration) Stream[I]
  • func Map[F, T any](source Stream[F], mapper Mapper[F, T]) Stream[T]
  • func Null[I any]() Stream[I]
  • func Pour[I any](slice ...I) Stream[I]
  • func Repeat[I any](item I, times int) Stream[I]
  • func Select[I any](source Stream[I], selector Selector[I]) Stream[I]
• type StreamFunc
  • func (f StreamFunc[I]) Next(ctx context.Context) (item I, err error)

Variables

var ErrEnd = errors.New("end")

ErrEnd is the error returned by by the Next method of streams when there are no more items in the stream.

Functions

func Collect

func Collect[I any](ctx context.Context, stream Stream[I]) (slice []I, err error)

Collect collects all the items in the given stream and returns an slice containing them.

Types

type Mapper

type Mapper[F, T any] func(context.Context, F) (T, error)

Mapper is a function that transforms one object into another.

type Selector

type Selector[I any] func(context.Context, I) (bool, error)

Selector is a function that filters element of a stream.

type Sizer

type Sizer interface {
	// Returns the number of items still available in the stream.
	Size(ctx context.Context) (size int, err error)
}

Sizer is an interface that can optionally be implemented by streams that know what is their size. Some functions, for example the Slice function that builds an slice from a stream, can take advantage of this to allocate the required space in advance.

type Stream

type Stream[I any] interface {
	// Next resturns the next item from the stream. Returns the EOS error if there are
	// no more items. Other errors may also be returned. For example, if the stream is backed
	// by a database table the database connection may fail and generate an error.
	Next(ctx context.Context) (item I, err error)
}

Stream represents a stream of items.

func Delay

func Delay[I any](source Stream[I], delay time.Duration) Stream[I]

Delay creates a new stream that returns the same items than the given source, but with an additional delay for each item. This is intended for tests and there is usually no reason to use in production code.

func Map

func Map[F, T any](source Stream[F], mapper Mapper[F, T]) Stream[T]

Map creates a stream that contains the result of transforming the objects of the given stream with a mapper. Note that the actual calls to the mapper will not happen when this function is called, they will happen only when the stream is eventually consumed.

func Null

func Null[I any]() Stream[I]

Null creates a new stream that is empty.

func Pour

func Pour[I any](slice ...I) Stream[I]

Pour creates a stream that contains the items in the given slice.

func Repeat

func Repeat[I any](item I, times int) Stream[I]

Repeat creates a stream that repeats the same item multiple times.

func Select

func Select[I any](source Stream[I], selector Selector[I]) Stream[I]

Select creates a new stream that only contains the items of the source stream that return true for the given selector. Note that the actual calls to the select will not happen when this function is called, they will happen only when the stream is eventually consumed.

type StreamFunc

type StreamFunc[I any] func(context.Context) (I, error)

StreamFunc creates an implementation of the Stream interface using the given function.

func (StreamFunc[I]) Next

func (f StreamFunc[I]) Next(ctx context.Context) (item I, err error)