machine

May 11, 2022

Changes in this version

type Applicative

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

type Builder

+ Combine func(x Combiner[T]) Builder[T]

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

+ OutputTo func(x chan []T)

+ Window func(x Window[T]) Builder[T]

+ type Combiner func(payload []T) T

+ func (x Combiner[T]) Component(output Edge[T]) Vertex[T]

type Comparator

+ func (x Comparator[T]) Component(output Edge[T]) Vertex[T]

+ type Component interface

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

type Edge

+ func AsEdge(c chan []T) Edge[T]

+ Input func(payload ...T)

+ OutputTo func(ctx context.Context, channel chan []T)

+ type Filter func(d T) FilterResult

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

+ type FilterResult uint8

+ const FilterBoth

+ const FilterLeft

+ const FilterRight

+ func Duplicate(d T) FilterResult

+ type Identifiable interface

+ ID func() string

type Option

+ MaxParallel int

+ PanicHandler func(err error, payload ...T)

+ Telemetry Telemetry[T]

type Remover

+ func (x Remover[T]) Component(output Edge[T]) Vertex[T]

+ type Span interface

+ RecordError func(error)

+ RecordPayload func(payload ...T)

+ SpanEnd func()

type Stream

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

+ Clear func()

+ Consume func(input ...chan []T)

+ Start func(ctx context.Context) error

+ StartWith func(ctx context.Context, input ...chan []T) error

+ type Telemetry interface

+ Duration func(time.Duration)

+ IncrementErrorCount func()

+ IncrementPayloadCount func()

+ PayloadSize func(int64)

+ StartSpan func() Span[T]

type Vertex

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

+ type Window func(payload []T) []T

+ func (x Window[T]) Component(output Edge[T]) Vertex[T]

Aug 17, 2021

Changes in this version

type Option

+ Validators map[string]ForkRule

Jul 27, 2021

Changes in this version

type HTTPStream

+ InjectionHandlers func() map[string]fiber.Handler

type Stream

+ VertexIDs func() []string

Jul 27, 2021

Changes in this version

+ type Edge interface

+ Next func(payload ...*Packet)

+ Send func(ctx context.Context, channel chan []*Packet)

+ type EdgeProvider interface

+ New func(options *Option) Edge

type Option

+ Provider EdgeProvider

Jul 20, 2021

Changes in this version

type Builder

+ Remove func(id string, x Remover) Builder

+ Sort func(id string, x Comparator) Builder

+ type Comparator func(a data.Data, b data.Data) int

+ type Remover func(index int, d data.Data) bool

Jul 19, 2021

Changes in this version

type HTTPStream

+ func LoadHTTP(serialization *StreamSerialization) (HTTPStream, error)

Jul 16, 2021

Changes in this version

+ type Cluster interface

+ Join func(id string, callback InjectionCallback) error

+ Leave func(id string) error

+ Write func(logs ...*Log)

+ type Error struct

+ Err error

+ Packets []*Packet

+ StreamID string

+ Time time.Time

+ VertexID string

+ VertexType string

+ func (e *Error) Error() string

+ type HTTPStream interface

+ Handler func() fiber.Handler

+ func NewHTTPStream(id string, opts ...*Option) HTTPStream

+ func NewWebsocketStream(id string, opts ...*Option) HTTPStream

type Packet

+ Errors map[string]error

type Stream

+ func Load(serialization *StreamSerialization) (Stream, error)

+ func NewSubscriptionStream(id string, sub Subscription, interval time.Duration, opts ...*Option) Stream

+ Errors func() chan error

+ InjectionCallback func(ctx context.Context) func(logs ...*Log)

type StreamSerialization

+ Options []*Option

Mar 24, 2021

Changes in this version

type Pipe

+ func NewPipeWithFiber(id string, logger *logrus.Logger, store LogStore, app *fiber.App) *Pipe

Mar 23, 2021

Changes in this version

type Pipe

+ func (pipe *Pipe) Add(method, path string, handlers ...func(*fiber.Ctx) error) fiber.Router

Mar 20, 2021

Changes in this version

type Pipe

+ func (pipe *Pipe) StreamWebsocket(id string, opts ...*Option) Builder

Mar 19, 2021

Changes in this version

type StreamSerialization

+ func (s *StreamSerialization) MarshalJSON() ([]byte, error)

+ func (s *StreamSerialization) UnmarshalJSON(data []byte) error

Mar 18, 2021

Changes in this version

type Builder

+ Publish func(id string, s Publisher, options ...*Option)

type LoopBuilder

+ Publish func(id string, s Publisher, options ...*Option)

+ type Publisher interface

+ Send func([]Data) error

Mar 7, 2021

Changes in this version

type PluginDefinition

+ Attributes map[string]interface{}

Mar 6, 2021

Changes in this version

+ func RegisterPluginProvider(name string, p PluginProvider)

+ type PluginDefinition struct

+ Payload string

+ Symbol string

+ Type string

+ type PluginProvider interface

+ Load func(*PluginDefinition) (interface{}, error)

+ type ProviderDefinitions struct

+ Plugins map[string]*PluginDefinition

+ func (pd *ProviderDefinitions) Load() error

+ type StreamSerialization struct

+ Interval time.Duration

+ Type string

+ func (s *StreamSerialization) MarshalYAML() (interface{}, error)

+ func (s *StreamSerialization) UnmarshalYAML(unmarshal func(interface{}) error) error

+ type VertexSerialization struct

+ Attributes map[string]interface{}

+ ID string

+ Options []*Option

+ Provider string

Feb 6, 2021

Changes in this version

type LoopBuilder

+ Transmit func(id string, s Sender, options ...*Option)

Feb 4, 2021

Changes in this version

type Builder

+ Loop func(id string, x Fork, options ...*Option) (loop LoopBuilder, out Builder)

+ type LoopBuilder interface

+ Done func()

+ FoldLeft func(id string, f Fold, options ...*Option) LoopBuilder

+ FoldRight func(id string, f Fold, options ...*Option) LoopBuilder

+ Fork func(id string, f Fork, options ...*Option) (LoopBuilder, LoopBuilder)

+ Loop func(id string, x Fork, options ...*Option) (loop LoopBuilder, out LoopBuilder)

+ Map func(id string, a Applicative, options ...*Option) LoopBuilder

Jan 1, 2021

Changes in this version

+ type DebugInfo struct

+ End time.Time

+ ID string

+ Snapshot Data

+ Start time.Time

type Option

+ Debug *bool

type Packet

+ Snapshots []*DebugInfo

Dec 31, 2020

Changes in this version

type Data

+ func (d *Data) As(i interface{}) error

type Packet

+ func (p *Packet) As(i interface{}) error

Dec 6, 2020

Changes in this version

type Option

+ DeepCopy *bool

Nov 24, 2020

Changes in this version

type Option

+ TraceID *bool

Nov 23, 2020

Changes in this version

type Pipe

+ func (pipe *Pipe) Load(lc *Serialization) error

+ type Serialization struct

+ ID string

+ Interval time.Duration

+ Left *Serialization

+ Next *Serialization

+ Options []*Option

+ Right *Serialization

+ Script string

+ Symbol string

+ To string

+ Type string

Nov 18, 2020

Changes in this version

type Pipe

+ func (pipe *Pipe) Stream(stream Stream) Builder

Nov 17, 2020

Changes in this version

+ type HealthInfo struct

+ LastPayload time.Time

+ StreamID string

+ type InjectionCallback func(logs ...*Log)

+ type Log struct

+ OwnerID string

+ Packet *Packet

+ State string

+ StreamID string

+ VertexID string

+ VertexType string

+ When time.Time

+ type LogStore interface

+ Join func(id string, callback InjectionCallback, streamIDs ...string) error

+ Leave func(id string) error

+ Write func(logs ...*Log)

+ type Logger interface

+ Error func(...interface{})

+ Info func(...interface{})

+ type Pipe struct

+ func NewPipe(id string, logger Logger, store LogStore, config ...fiber.Config) *Pipe

+ func (pipe *Pipe) Run(ctx context.Context, port string, gracePeriod time.Duration) error

+ func (pipe *Pipe) StreamHTTP(id string, opts ...*Option) Builder

+ func (pipe *Pipe) StreamSubscription(id string, sub Subscription, interval time.Duration, opts ...*Option) Builder

+ func (pipe *Pipe) Use(args ...interface{})

+ type Subscription interface

+ Close func() error

+ Read func(ctx context.Context) []Data

Nov 15, 2020

Changes in this version

type Option

+ Injectable *bool

Nov 9, 2020

Changes in this version

type Builder

+ Link func(id, target string)

Nov 5, 2020

Changes in this version

type Builder

+ FoldLeft func(id string, f Fold) Builder

+ FoldRight func(id string, f Fold) Builder

+ Fork func(id string, f Fork) (Builder, Builder)

+ Map func(id string, a Applicative) Builder

+ type Fold func(Data, Data) Data

+ type Fork func(list []*Packet) (a, b []*Packet)

+ var ForkDuplicate Fork = func(payload []*Packet) (a, b []*Packet) { ... }

+ var ForkError Fork = func(payload []*Packet) (s, f []*Packet) { ... }

+ type ForkRule func(Data) bool

+ func (r ForkRule) Handler(payload []*Packet) (t, f []*Packet)

+ type Stream interface

+ Builder func() Builder

+ ID func() string

+ Inject func(ctx context.Context, events map[string][]*Packet)

+ Run func(ctx context.Context, recorders ...recorder) error

+ func NewStream(id string, retriever Retriever, options ...*Option) Stream

Nov 1, 2020

Changes in this version

+ type Data typed.Typed

Nov 1, 2020

Changes in this version

type Builder

+ func (m *Builder) Split(r *Splitter) *Builder

+ func (m *Builder) Transmit(t *Transmission) *Builder

type Splitter

+ func (m *Splitter) SplitLeft(left *Splitter) *Splitter

+ func (m *Splitter) SplitRight(right *Splitter) *Splitter

+ func (m *Splitter) TransmitLeft(t *Transmission) *Splitter

+ func (m *Splitter) TransmitRight(t *Transmission) *Splitter

type Vertex

+ func (m *Vertex) Split(r *Splitter) *Vertex

+ func (m *Vertex) Transmit(t *Transmission) *Vertex

Nov 1, 2020

Changes in this version

+ type Applicative func(typed.Typed) error

+ type Retriever func(context.Context) chan []typed.Typed

+ type Sender func([]typed.Typed) error

+ type SplitHandler func(list []*Packet) (a, b []*Packet)

+ var SplitDuplicate SplitHandler = func(payload []*Packet) (a, b []*Packet) { ... }

+ var SplitError SplitHandler = func(payload []*Packet) (s, f []*Packet) { ... }

+ type SplitRule func(typed.Typed) bool

+ func (r SplitRule) Handler(payload []*Packet) (t, f []*Packet)

+ type Splitter struct

+ func NewSplitter(id string, s SplitHandler) *Splitter

+ func (m *Splitter) RouteLeft(left *Splitter) *Splitter

+ func (m *Splitter) RouteRight(right *Splitter) *Splitter

+ func (m *Splitter) TerminateLeft(t *Transmission) *Splitter

+ func (m *Splitter) TerminateRight(t *Transmission) *Splitter

+ func (m *Splitter) ThenLeft(left *Vertex) *Splitter

+ func (m *Splitter) ThenRight(right *Vertex) *Splitter

+ type Transmission struct

+ func NewTransmission(id string, s Sender) *Transmission

+ type Vertex struct

+ func (m *Vertex) Route(r *Splitter) *Vertex

+ func (m *Vertex) Terminate(t *Transmission) *Vertex

+ func (m *Vertex) Then(v *Vertex) *Vertex

Oct 31, 2020

Changes in this version

type Option

+ Metrics *bool

+ Span *bool

Oct 31, 2020

Changes in this version

type Builder

+ func (m *Builder) ID() string

+ func (m *Builder) Inject(ctx context.Context, events map[string][]*Packet)

+ type Option struct

+ BufferSize *int

+ FIFO *bool

+ Idempotent *bool

Oct 30, 2020

Changes in this version

type Builder

+ func (m *Builder) Run(ctx context.Context, bufferSize int, ...) error

Oct 21, 2020

Changes in this version

+ type Builder struct

+ func New(id, name string, fifo bool, i Initium) *Builder

+ func (m *Builder) Build(bufferSize int, recorders ...func(string, string, []*Packet)) *Machine

+ func (m *Builder) Route(r *RouterBuilder) *Builder

+ func (m *Builder) Terminate(t *TerminationBuilder) *Builder

+ func (m *Builder) Then(v *VertexBuilder) *Builder

+ type Initium func(context.Context) chan []map[string]interface

+ type Machine struct

+ func (m *Machine) ID() string

+ func (m *Machine) Inject(logs map[string][]*Packet)

+ func (m *Machine) Run(ctx context.Context) error

+ type Packet struct

+ Data map[string]interface{}

+ Error error

+ ID string

+ type Processus func(map[string]interface{}) error

+ type RouteHandler func(list []*Packet) (t, f []*Packet)

+ type RouterBuilder struct

+ func NewRouter(id, name string, fifo bool, r RouteHandler) *RouterBuilder

+ func (m *RouterBuilder) RouteLeft(left *RouterBuilder) *RouterBuilder

+ func (m *RouterBuilder) RouteRight(right *RouterBuilder) *RouterBuilder

+ func (m *RouterBuilder) TerminateLeft(t *TerminationBuilder) *RouterBuilder

+ func (m *RouterBuilder) TerminateRight(t *TerminationBuilder) *RouterBuilder

+ func (m *RouterBuilder) ThenLeft(left *VertexBuilder) *RouterBuilder

+ func (m *RouterBuilder) ThenRight(right *VertexBuilder) *RouterBuilder

+ type RouterDuplicate struct

+ func (d RouterDuplicate) Handler(payload []*Packet) (a, b []*Packet)

+ type RouterError struct

+ func (e RouterError) Handler(payload []*Packet) (s, f []*Packet)

+ type RouterRule func(map[string]interface{}) bool

+ func (r RouterRule) Handler(payload []*Packet) (t, f []*Packet)

+ type TerminationBuilder struct

+ func NewTermination(id, name string, fifo bool, t Terminus) *TerminationBuilder

+ type Terminus func([]map[string]interface{}) error

+ type VertexBuilder struct

+ func NewVertex(id, name string, fifo bool, p Processus) *VertexBuilder

+ func (m *VertexBuilder) Route(r *RouterBuilder) *VertexBuilder

+ func (m *VertexBuilder) Terminate(t *TerminationBuilder) *VertexBuilder

+ func (m *VertexBuilder) Then(v *VertexBuilder) *VertexBuilder