streams

README

🏄♀ Streams

A teeny-tiny package to create stream processing workloads.

Docs

You can find the documentation hosted on godoc.org.

Examples

See the examples directory for more.

Operators

• Do: Execute a function for each element in the stream.
• Filter: Filter elements from the stream.
• Map: Transform elements in the stream.
• Reduce: Reduce elements in the stream.
• Split: Split the stream into multiple streams.
• Merge: Merge multiple streams into one.
• FlatMap: Transform elements in the stream into multiple elements.
• Skip: Skip elements in the stream.

License

Apache 2.0

Documentation

Overview

Package streams provides a set of functions to work with event streams. The package is designed to be used with Kafka, NATS, and other message brokers.

Index

• func Pipe(stream Streamable, rev Receivable)
• type Do
  • func NewDo[T any](fn DoFunc[T]) *Do[T]
  • func (d *Do[T]) In() chan<- any
  • func (d *Do[T]) Out() <-chan any
  • func (d *Do[T]) Pipe(c Operatable) Operatable
  • func (d *Do[T]) To(sink Sinkable)
• type DoFunc
• type Filter
  • func NewFilter[T any](fn FilterPredicate[T]) *Filter[T]
  • func (f *Filter[T]) In() chan<- any
  • func (f *Filter[T]) Out() <-chan any
  • func (f *Filter[T]) Pipe(c Operatable) Operatable
  • func (f *Filter[T]) To(sink Sinkable)
• type FilterPredicate
• type FlatMap
  • func NewFlatMap[T, R any](fn FlatMapFunc[T, R]) *FlatMap[T, R]
  • func (f *FlatMap[T, R]) In() chan<- any
  • func (f *FlatMap[T, R]) Out() <-chan any
  • func (f *FlatMap[T, R]) Pipe(c Operatable) Operatable
  • func (f *FlatMap[T, R]) To(sink Sinkable)
• type FlatMapFunc
• type Log
  • func NewLog(fn logx.LogFunc) *Log
  • func (l *Log) In() chan<- any
  • func (l *Log) Out() <-chan any
  • func (l *Log) Pipe(c Operatable) Operatable
  • func (l *Log) To(sink Sinkable)
• type Map
  • func NewMap[T, R any](fn MapFunc[T, R]) *Map[T, R]
  • func (m *Map[T, R]) In() chan<- any
  • func (m *Map[T, R]) Out() <-chan any
  • func (m *Map[T, R]) Pipe(c Operatable) Operatable
  • func (m *Map[T, R]) To(sink Sinkable)
• type MapFunc
• type Node
  • func NewNode(name string) Node
• type Operatable
  • func FanOut(in Streamable, num int) []Operatable
  • func Flatten[T any]() Operatable
  • func Merge(in ...Streamable) Operatable
  • func Split[T any](in Streamable, predicate FilterPredicate[T]) [2]Operatable
• type PassThrough
  • func NewPassThrough() *PassThrough
  • func (p *PassThrough) In() chan<- any
  • func (p *PassThrough) Out() <-chan any
  • func (p *PassThrough) Pipe(c Operatable) Operatable
  • func (p *PassThrough) To(sink Sinkable)
• type Receivable
• type Reduce
  • func NewReduce[T any](fn ReduceFunc[T]) *Reduce[T]
  • func (r *Reduce[T]) In() chan<- any
  • func (r *Reduce[T]) Out() <-chan any
  • func (r *Reduce[T]) Pipe(c Operatable) Operatable
  • func (r *Reduce[T]) To(sink Sinkable)
• type ReduceFunc
• type Sinkable
• type Skip
  • func NewSkip(n int) *Skip
  • func (s *Skip) In() chan<- any
  • func (s *Skip) Out() <-chan any
  • func (s *Skip) Pipe(c Operatable) Operatable
  • func (s *Skip) To(sink Sinkable)
• type Sourceable
• type Streamable
• type Topology
  • func NewTopology(root Node) Topology

Functions

func Pipe

func Pipe(stream Streamable, rev Receivable)

Pipe pipes the output channel to the input channel.

Types

type Do

type Do[T any] struct {
	// contains filtered or unexported fields
}

Do takes one element and executes a function on it.

func NewDo

func NewDo[T any](fn DoFunc[T]) *Do[T]

NewDo creates a new Do.

func (*Do[T]) In

func (d *Do[T]) In() chan<- any

In returns the input channel.

func (*Do[T]) Out

func (d *Do[T]) Out() <-chan any

Out returns the output channel.

func (*Do[T]) Pipe

func (d *Do[T]) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*Do[T]) To

func (d *Do[T]) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type DoFunc

type DoFunc[T any] func(T) error

DoFunc is a function that executes on the element.

type Filter

type Filter[T any] struct {
	// contains filtered or unexported fields
}

Filter filters an incoming element using a filter predicate.

func NewFilter

func NewFilter[T any](fn FilterPredicate[T]) *Filter[T]

NewFilter returns a new operator on filters.

func (*Filter[T]) In

func (f *Filter[T]) In() chan<- any

In returns the input channel.

func (*Filter[T]) Out

func (f *Filter[T]) Out() <-chan any

Out returns the output channel.

func (*Filter[T]) Pipe

func (f *Filter[T]) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*Filter[T]) To

func (f *Filter[T]) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type FilterPredicate

type FilterPredicate[T any] func(T) bool

FilterPredicate represents a filter predicate.

type FlatMap

type FlatMap[T, R any] struct {
	// contains filtered or unexported fields
}

FlatMap takes one element and produces a new element of the same type.

func NewFlatMap

func NewFlatMap[T, R any](fn FlatMapFunc[T, R]) *FlatMap[T, R]

NewFlatMap returns a new operator on maps.

func (*FlatMap[T, R]) In

func (f *FlatMap[T, R]) In() chan<- any

In returns the input channel.

func (*FlatMap[T, R]) Out

func (f *FlatMap[T, R]) Out() <-chan any

Out returns the output channel.

func (*FlatMap[T, R]) Pipe

func (f *FlatMap[T, R]) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*FlatMap[T, R]) To

func (f *FlatMap[T, R]) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type FlatMapFunc

type FlatMapFunc[T, R any] func(T) []R

FlatMap takes one element and produces a new element of the same type.

type Log

type Log struct {
	// contains filtered or unexported fields
}

Log passes through an incoming element.

func NewLog

func NewLog(fn logx.LogFunc) *Log

NewLog returns a new operator to log elements.

func (*Log) In

func (l *Log) In() chan<- any

In returns the input channel.

func (*Log) Out

func (l *Log) Out() <-chan any

Out returns the output channel.

func (*Log) Pipe

func (l *Log) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*Log) To

func (l *Log) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type Map

type Map[T, R any] struct {
	// contains filtered or unexported fields
}

Map takes one element and produces a new element of the same type.

func NewMap

func NewMap[T, R any](fn MapFunc[T, R]) *Map[T, R]

NewMap returns a new operator on maps.

func (*Map[T, R]) In

func (m *Map[T, R]) In() chan<- any

In returns the input channel.

func (*Map[T, R]) Out

func (m *Map[T, R]) Out() <-chan any

Out returns the output channel.

func (*Map[T, R]) Pipe

func (m *Map[T, R]) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*Map[T, R]) To

func (m *Map[T, R]) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type MapFunc

type MapFunc[T, R any] func(T) R

MapFunc is a function that takes a key and a value and returns a new value.

type Node

type Node interface {
	// AddChildren adds children to a node.
	AddChild(nodes ...Node)
	// Children returns the children of a node.
	Children() []Node
	// Name returns the name of a node.
	Name(names ...string) string
}

Node is a node in a topology.

func NewNode

func NewNode(name string) Node

NewNode is a constructor for a new node in the topology.

type Operatable

type Operatable interface {
	Streamable
	Receivable
	// To streams data to the sink and waits for it to complete.
	To(sink Sinkable)
}

Operatable is a Operatable interface.

func FanOut

func FanOut(in Streamable, num int) []Operatable

FanOut fans out a stream to multiple streams.

func Flatten

func Flatten[T any]() Operatable

Flatten creates a flatten stream.

func Merge

func Merge(in ...Streamable) Operatable

Merge merges multiple streams into one.

func Split

func Split[T any](in Streamable, predicate FilterPredicate[T]) [2]Operatable

Split splits a stream in two based on a predicate.

type PassThrough

type PassThrough struct {
	// contains filtered or unexported fields
}

PassThrough passes through an incoming element.

func NewPassThrough

func NewPassThrough() *PassThrough

NewPassThrough returns a new operator on pass-throughs.

func (*PassThrough) In

func (p *PassThrough) In() chan<- any

In returns the input channel.

func (*PassThrough) Out

func (p *PassThrough) Out() <-chan any

Out returns the output channel.

func (*PassThrough) Pipe

func (p *PassThrough) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*PassThrough) To

func (p *PassThrough) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type Receivable

type Receivable interface {
	// In returns the input channel.
	In() chan<- any
}

Receivable is a receivable interface.

type Reduce

type Reduce[T any] struct {
	// contains filtered or unexported fields
}

Reduce takes the current element and the latest reduced value and produces a new reduced value.

func NewReduce

func NewReduce[T any](fn ReduceFunc[T]) *Reduce[T]

NewReduce returns a new operator on reduces.

func (*Reduce[T]) In

func (r *Reduce[T]) In() chan<- any

In returns the input channel.

func (*Reduce[T]) Out

func (r *Reduce[T]) Out() <-chan any

Out returns the output channel.

func (*Reduce[T]) Pipe

func (r *Reduce[T]) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*Reduce[T]) To

func (r *Reduce[T]) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type ReduceFunc

type ReduceFunc[T any] func(T, T) T

ReduceFunc combines the current element with the latest reduced value.

type Sinkable

type Sinkable interface {
	Receivable
	// Wait waits for the sink to complete.
	Wait()
	// Error returns the error.
	Error() error
}

Sinkable is a sinkable interface.

type Skip

type Skip struct {
	// contains filtered or unexported fields
}

Skip skips the first n elements.

func NewSkip

func NewSkip(n int) *Skip

NewSkip returns a new operator on skips.

func (*Skip) In

func (s *Skip) In() chan<- any

In returns the input channel.

func (*Skip) Out

func (s *Skip) Out() <-chan any

Out returns the output channel.

func (*Skip) Pipe

func (s *Skip) Pipe(c Operatable) Operatable

Pipe pipes the output channel to the input channel.

func (*Skip) To

func (s *Skip) To(sink Sinkable)

To streams data to the sink and waits for it to complete.

type Sourceable

type Sourceable interface {
	Streamable
	// Error returns the error.
	Error() error
}

Sourceable is a sourceable interface.

type Streamable

type Streamable interface {
	// Out returns the output channel.
	Out() <-chan any
	// Pipe pipes the output channel to the input channel.
	Pipe(Operatable) Operatable
}

Streamable is a streamable interface.

type Topology

type Topology interface {
	// Root returns the root node of a topology.
	Root() Node
}

Topology is a graph of nodes.

func NewTopology

func NewTopology(root Node) Topology

NewTopology is a constructor for Topology.