machine

Jul 17, 2023

Changes in this version

type Edge

+ Output func() chan T

+ type Machine interface

+ And func(x ...Filter[T]) (Machine[T], Machine[T])

+ AsEdge func() Edge[T]

+ Distribute func(Edge[T]) Machine[T]

+ Drop func()

+ Duplicate func() (Machine[T], Machine[T])

+ If func(f Filter[T]) (Machine[T], Machine[T])

+ Memoize func(x Monad[Monad[T]], index func(T) string) Machine[T]

+ Name func() string

+ Or func(x ...Filter[T]) (Machine[T], Machine[T])

+ Output func() chan T

+ Recurse func(x Monad[Monad[T]]) Machine[T]

+ Select func(fns ...Filter[T]) []Machine[T]

+ Then func(a Monad[T]) Machine[T]

+ While func(x Filter[T]) (loop, out Machine[T])

+ type Monad func(d T) T

Apr 20, 2023

Changes in this version

type Builder

+ When func(fns ...Filter[T]) []Builder[T]

Nov 8, 2022

Changes in this version

+ type Transform func(d Applicative[T]) Applicative[T]

+ func (x Transform[T]) Component(output chan T) Vertex[T]

Nov 8, 2022

Changes in this version

+ type BaseFn func(d T) T

+ type BaseFnTransform func(d BaseFn[T]) BaseFn[T]

+ func (x BaseFnTransform[T]) Component(output chan T) Vertex[T]

type Builder

+ Y func(x BaseFnTransform[T]) Builder[T]

Oct 31, 2022

Changes in this version

type Option

+ DeepCopyBetweenVerticies bool

Oct 30, 2022

Changes in this version

+ type Applicative func(d T) T

+ func (x Applicative[T]) Component(output chan T) Vertex[T]

+ type Builder interface

+ And func(x ...Test[T]) (Builder[T], Builder[T])

+ Distribute func(Edge[T]) Builder[T]

+ Drop func()

+ Duplicate func() (Builder[T], Builder[T])

+ Filter func(f Filter[T]) (Builder[T], Builder[T])

+ Loop func(x Filter[T]) (loop, out Builder[T])

+ Or func(x ...Test[T]) (Builder[T], Builder[T])

+ OutputTo func(x chan T)

+ Then func(a Applicative[T]) Builder[T]

+ type Component interface

+ Component func(e chan T) Vertex[T]

+ type Edge interface

+ ReceiveOn func(ctx context.Context, channel chan T)

+ Send func(payload T)

+ type Filter func(d T) bool

+ func (x Filter[T]) Component(left, right chan T, _ *Option[T]) Vertex[T]

+ type Option struct

+ BufferSize int

+ DeepCopy func(T) T

+ FIFO bool

+ PanicHandler func(err error, payload T)

+ Telemetry Telemetry[T]

+ type Stream interface

+ Builder func() Builder[T]

+ Start func(ctx context.Context, input chan T) error

+ func New(name string, options *Option[T]) Stream[T]

+ type Telemetry interface

+ Duration func(vertexName string, duration time.Duration)

+ IncrementErrorCount func(vertexName string)

+ IncrementPayloadCount func(vertexName string)

+ RecordError func(vertexName string, payload T, err error)

+ RecordPayload func(vertexName string, payload T)

+ type Test func(d T) (T, error)

+ func (x Test[T]) Component(left, right chan T, _ *Option[T]) Vertex[T]

+ type Vertex func(payload T)

+ func (x Vertex[T]) Run(ctx context.Context, name string, channel chan T, option *Option[T])