flow

Jan 14, 2024

Changes in this version

+ type Batch struct

+ func NewBatch[T any](maxBatchSize int, timeInterval time.Duration) *Batch[T]

+ func (b *Batch[T]) In() chan<- any

+ func (b *Batch[T]) Out() <-chan any

+ func (b *Batch[T]) To(sink streams.Sink)

+ func (b *Batch[T]) Via(flow streams.Flow) streams.Flow

type SlidingWindow

+ func NewSlidingWindowWithExtractor[T any](windowSize time.Duration, slidingInterval time.Duration, ...) *SlidingWindow[T]

Oct 17, 2022

Changes in this version

+ func Flatten(parallelism uint) streams.Flow

+ func RoundRobin(outlet streams.Outlet, magnitude int) []streams.Flow

+ type Reduce struct

+ func NewReduce(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 streams.Sink)

+ func (r *Reduce[T]) Via(flow streams.Flow) streams.Flow

+ type ReduceFunction func(T, T) T

Nov 27, 2021

Changes in this version

+ type FilterPredicate func(interface{}) bool

+ type FlatMapFunction func(interface{}) []interface

+ type MapFunction func(interface{}) interface

+ type SessionWindow struct

+ func NewSessionWindow(inactivityGap time.Duration) *SessionWindow

+ func (sw *SessionWindow) In() chan<- interface{}

+ func (sw *SessionWindow) Out() <-chan interface{}

+ func (sw *SessionWindow) To(sink streams.Sink)

+ func (sw *SessionWindow) Via(flow streams.Flow) streams.Flow

Dec 19, 2020

Changes in this version

type SlidingWindow

+ func NewSlidingWindowWithTSExtractor(size time.Duration, slide time.Duration, ...) *SlidingWindow

Aug 4, 2019

Changes in this version

+ func DoStream(outlet streams.Outlet, inlet streams.Inlet)

+ func FanOut(outlet streams.Outlet, magnitude int) []streams.Flow

+ func Merge(outlets ...streams.Flow) streams.Flow

+ func Split(outlet streams.Outlet, cond func(interface{}) bool) [2]streams.Flow

+ type Filter struct

+ FilterF FilterFunc

+ func NewFilter(f FilterFunc, parallelism uint) *Filter

+ func (f *Filter) In() chan<- interface{}

+ func (f *Filter) Out() <-chan interface{}

+ func (f *Filter) To(sink streams.Sink)

+ func (f *Filter) Via(flow streams.Flow) streams.Flow

+ type FilterFunc func(interface{}) bool

+ type FlatMap struct

+ FlatMapF FlatMapFunc

+ func NewFlatMap(f FlatMapFunc, parallelism uint) *FlatMap

+ func (fm *FlatMap) In() chan<- interface{}

+ func (fm *FlatMap) Out() <-chan interface{}

+ func (fm *FlatMap) To(sink streams.Sink)

+ func (fm *FlatMap) Via(flow streams.Flow) streams.Flow

+ type FlatMapFunc func(interface{}) []interface

+ type Item struct

+ Msg interface{}

+ func NewItem(msg interface{}, epoch int64, index int) *Item

+ type Map struct

+ MapF MapFunc

+ func NewMap(f MapFunc, parallelism uint) *Map

+ func (m *Map) In() chan<- interface{}

+ func (m *Map) Out() <-chan interface{}

+ func (m *Map) To(sink streams.Sink)

+ func (m *Map) Via(flow streams.Flow) streams.Flow

+ type MapFunc func(interface{}) interface

+ type PassThrough struct

+ func NewPassThrough() *PassThrough

+ func (pt *PassThrough) In() chan<- interface{}

+ func (pt *PassThrough) Out() <-chan interface{}

+ func (pt *PassThrough) To(sink streams.Sink)

+ func (pt *PassThrough) Via(flow streams.Flow) streams.Flow

+ type PriorityQueue []*Item

+ func (pq *PriorityQueue) Head() *Item

+ func (pq *PriorityQueue) Pop() interface{}

+ func (pq *PriorityQueue) Push(x interface{})

+ func (pq *PriorityQueue) Update(item *Item, newEpoch int64)

+ func (pq PriorityQueue) Len() int

+ func (pq PriorityQueue) Less(i, j int) bool

+ func (pq PriorityQueue) Slice(start, end int) PriorityQueue

+ func (pq PriorityQueue) Swap(i, j int)

+ type SlidingWindow struct

+ func NewSlidingWindow(size time.Duration, slide time.Duration) *SlidingWindow

+ func NewSlidingWindowWithTsExtractor(size time.Duration, slide time.Duration, ...) *SlidingWindow

+ func (sw *SlidingWindow) In() chan<- interface{}

+ func (sw *SlidingWindow) Out() <-chan interface{}

+ func (sw *SlidingWindow) To(sink streams.Sink)

+ func (sw *SlidingWindow) Via(flow streams.Flow) streams.Flow

+ type ThrottleMode int8

+ const Backpressure

+ const Discard

+ type Throttler struct

+ func NewThrottler(elements uint, per time.Duration, buffer uint, mode ThrottleMode) *Throttler

+ func (th *Throttler) In() chan<- interface{}

+ func (th *Throttler) Out() <-chan interface{}

+ func (th *Throttler) To(sink streams.Sink)

+ func (th *Throttler) Via(flow streams.Flow) streams.Flow

+ type TumblingWindow struct

+ func NewTumblingWindow(size time.Duration) *TumblingWindow

+ func (tw *TumblingWindow) In() chan<- interface{}

+ func (tw *TumblingWindow) Out() <-chan interface{}

+ func (tw *TumblingWindow) To(sink streams.Sink)

+ func (tw *TumblingWindow) Via(flow streams.Flow) streams.Flow