gstream

type CursorNext ¶

type CursorNext[T any] func(ctx context.Context) (items []T, hasNext bool, err error)

type Filter ¶

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

Filter filters incoming elements using a filter predicate. If an element matches the predicate, the element is passed downstream. If not, the element is discarded.

in -- 1 -- 2 ---- 3 -- 4 ------ 5 --

[ -------- FilterPredicate -------- ]

out -- 1 -- 2 ------------------ 5 --

func NewFilter[T any](filterPredicate FilterPredicate[T], parallelism ...uint) *Filter[T]

func (f *Filter[T]) In() chan<- interface{}

func (f *Filter[T]) Out() <-chan interface{}

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

func (f *Filter[T]) Via(flow Transfer) Transfer

type FilterPredicate ¶

type FilterPredicate[T any] func(context.Context, T) (bool, error)

FilterPredicate represents a filter predicate (boolean-valued function).

type FlatMap ¶

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

FlatMap takes one element and produces zero, one, or more elements.

in -- 1 -- 2 ---- 3 -- 4 ------ 5 --

[ -------- FlatMapFunction -------- ]

out -- 1' - 2' -------- 4'- 4" - 5' -

func NewFlatMap[T, R any](flatMapFunction FlatMapFunction[T, R], parallelism ...uint) *FlatMap[T, R]

func (fm *FlatMap[T, R]) In() chan<- interface{}

func (fm *FlatMap[T, R]) Out() <-chan interface{}

func (fm *FlatMap[T, R]) To(sink Sink)

func (fm *FlatMap[T, R]) Via(flow Transfer) Transfer

type FlatMapFunction ¶

type FlatMapFunction[T, R any] func(context.Context, T) ([]R, error)

FlatMapFunction represents a FlatMap transformation function.

type FlowState ¶

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

func FlowStateWithContext(ctx context.Context) FlowState

func (bs *FlowState) Context() context.Context

func (bs *FlowState) Done()

func (bs *FlowState) HasStateErr() bool

func (bs *FlowState) SetStateErr(err error)

func (bs *FlowState) State() *state

type Inlet ¶

type Inlet interface {
	Stater
	In() chan<- interface{}
}

Inlet represents a type that exposes one open input.

type Map ¶

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

Map takes one element and produces one element.

in -- 1 -- 2 ---- 3 -- 4 ------ 5 --

[ ---------- MapFunction ---------- ]

out -- 1' - 2' --- 3' - 4' ----- 5' -

func NewMap[T, R any](mapFunction MapFunction[T, R], parallelism ...uint) *Map[T, R]

func (m *Map[T, R]) In() chan<- interface{}

func (m *Map[T, R]) Out() <-chan interface{}

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

func (m *Map[T, R]) Via(flow Transfer) Transfer

type MapFunction ¶

type MapFunction[T, R any] func(context.Context, T) (R, error)

MapFunction represents a Map transformation function.

type MemorySink ¶

type MemorySink[T any] interface {
	Sink
	Result() []T
}

func NewMemorySink[T any]() MemorySink[T]

type Optional ¶ added in v1.0.1

type Optional[T any] interface {
	Get() T
	IsPresent() bool
}

func NewEmptyOptional[T any]() Optional[T]

func NewOptional[T any](t T) Optional[T]

type Outlet ¶

type Outlet interface {
	Stater
	Out() <-chan interface{}
}

Outlet represents a type that exposes one open output.

type Reduce ¶

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

Reduce represents a “rolling” reduce on a data stream. Combines the current element with the last reduced value and emits the new value.

in -- 1 -- 2 ---- 3 -- 4 ------ 5 --

[ --------- ReduceFunction --------- ]

out -- 1 -- 2' --- 3' - 4' ----- 5' -

func NewReduce[T any](reduceFunction ReduceFunction[T]) *Reduce[T]

func (r *Reduce[T]) In() chan<- interface{}

func (r *Reduce[T]) Out() <-chan interface{}

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

func (r *Reduce[T]) Via(flow Transfer) Transfer

type ReduceFunction ¶

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

ReduceFunction combines the current element with the last reduced value.

type Sink ¶

type Sink interface {
	Inlet
	// contains filtered or unexported methods
}

Sink represents a set of stream processing steps that has one open input.

type Source ¶

type Source interface {
	Outlet
	Via(Transfer) Transfer
}

Source represents a set of stream processing steps that has one open output.

func NewDataStream(out Outlet) Source

func NewDataStreamOfCursor[T any](ctx context.Context, cursor func(ctx context.Context) CursorNext[T]) Source

func NewDataStreamOfSlice[T any](ctx context.Context, items []T) Source

type Stater ¶

type Stater interface {
	State() *state
	Context() context.Context

	SetStateErr(err error)
	// contains filtered or unexported methods
}

type StoreSink ¶ added in v1.0.1

type StoreSink[T any] interface {
	Sink
}

func NewStoreSink[T any](batch int, store func(ctx context.Context, ts []T) error) StoreSink[T]

type Transfer ¶

type Transfer interface {
	Inlet
	Outlet
	Via(Transfer) Transfer
	To(Sink)
}

Transfer represents a set of stream processing steps that has one open input and one open output.

func FanOut(outlet Outlet, magnitude int) []Transfer

func Merge(outlets ...Transfer) Transfer

func Split[T any](outlet Outlet, predicate func(T) bool) [2]Transfer