streams

package module

Go to main page

Versions in this module

v1

v1.1.0

Oct 17, 2018

Changes in this version

type Node

+ Processor func() Processor

type Pipe

+ func NewPipe(children []Pump) Pipe

type ProcessorNode

+ func (n *ProcessorNode) Processor() Processor

+ type Pump interface

+ Close func() error

+ Process func(*Message) error

+ func NewPump(node Node, pipe TimedPipe, errFn ErrorFunc) Pump

type SourceNode

+ func (n *SourceNode) Processor() Processor

+ type SourcePump interface

+ Close func() error

+ Stop func()

+ func NewSourcePump(name string, source Source, pumps []Pump, errFn ErrorFunc) SourcePump

+ type SourcePumps []SourcePump

+ func (p SourcePumps) StopAll()

+ type TimedPipe interface

+ Duration func() time.Duration

+ Reset func()

v1.0.0

Aug 31, 2018

Changes in this version

+ type BranchProcessor struct

+ func (p *BranchProcessor) Close() error

+ func (p *BranchProcessor) Process(msg *Message) error

+ func (p *BranchProcessor) WithPipe(pipe Pipe)

+ type ErrorFunc func(error)

+ type FilterProcessor struct

+ func (p *FilterProcessor) Close() error

+ func (p *FilterProcessor) Process(msg *Message) error

+ func (p *FilterProcessor) WithPipe(pipe Pipe)

+ type FlatMapProcessor struct

+ func (p *FlatMapProcessor) Close() error

+ func (p *FlatMapProcessor) Process(msg *Message) error

+ func (p *FlatMapProcessor) WithPipe(pipe Pipe)

+ type FlatMapper func(*Message) ([]*Message, error)

+ type MapProcessor struct

+ func (p *MapProcessor) Close() error

+ func (p *MapProcessor) Process(msg *Message) error

+ func (p *MapProcessor) WithPipe(pipe Pipe)

+ type Mapper func(*Message) (*Message, error)

+ type MergeProcessor struct

+ func (p *MergeProcessor) Close() error

+ func (p *MergeProcessor) Process(msg *Message) error

+ func (p *MergeProcessor) WithPipe(pipe Pipe)

+ type Message struct

+ Ctx context.Context

+ Key interface{}

+ Value interface{}

+ func NewMessage(k, v interface{}) *Message

+ func NewMessageWithContext(ctx context.Context, k, v interface{}) *Message

+ func (m *Message) Metadata() map[Source]interface{}

+ func (m *Message) WithMetadata(s Source, v interface{}) *Message

+ func (m Message) Empty() bool

+ type Node interface

+ AddChild func(n Node)

+ Children func() []Node

+ Close func() error

+ Name func() string

+ Process func(*Message) ([]NodeMessage, error)

+ WithPipe func(Pipe)

+ type NodeMessage struct

+ Msg *Message

+ Node Node

+ type Pipe interface

+ Commit func(*Message) error

+ Forward func(*Message) error

+ ForwardToChild func(*Message, int) error

+ Queue func() []NodeMessage

+ type Predicate func(*Message) (bool, error)

+ type PrintProcessor struct

+ func (p *PrintProcessor) Close() error

+ func (p *PrintProcessor) Process(msg *Message) error

+ func (p *PrintProcessor) WithPipe(pipe Pipe)

+ type Processor interface

+ Close func() error

+ Process func(*Message) error

+ WithPipe func(Pipe)

+ func NewBranchProcessor(fns []Predicate) Processor

+ func NewFilterProcessor(fn Predicate) Processor

+ func NewFlatMapProcessor(fn FlatMapper) Processor

+ func NewMapProcessor(fn Mapper) Processor

+ func NewMergeProcessor() Processor

+ func NewPrintProcessor() Processor

+ type ProcessorNode struct

+ func NewProcessorNode(name string, p Processor) *ProcessorNode

+ func (n *ProcessorNode) AddChild(node Node)

+ func (n *ProcessorNode) Children() []Node

+ func (n *ProcessorNode) Close() error

+ func (n *ProcessorNode) Name() string

+ func (n *ProcessorNode) Process(msg *Message) ([]NodeMessage, error)

+ func (n *ProcessorNode) WithPipe(pipe Pipe)

+ type ProcessorPipe struct

+ func NewProcessorPipe(node Node) *ProcessorPipe

+ func (p *ProcessorPipe) Commit(msg *Message) error

+ func (p *ProcessorPipe) Forward(msg *Message) error

+ func (p *ProcessorPipe) ForwardToChild(msg *Message, index int) error

+ func (p *ProcessorPipe) Queue() []NodeMessage

+ type Source interface

+ Close func() error

+ Commit func(interface{}) error

+ Consume func() (*Message, error)

+ type SourceNode struct

+ func NewSourceNode(name string) *SourceNode

+ func (n *SourceNode) AddChild(node Node)

+ func (n *SourceNode) Children() []Node

+ func (n *SourceNode) Close() error

+ func (n *SourceNode) Name() string

+ func (n *SourceNode) Process(msg *Message) ([]NodeMessage, error)

+ func (n *SourceNode) WithPipe(pipe Pipe)

+ type Stream struct

+ func (s *Stream) Branch(name string, preds ...Predicate) []*Stream

+ func (s *Stream) Filter(name string, pred Predicate) *Stream

+ func (s *Stream) FlatMap(name string, mapper FlatMapper) *Stream

+ func (s *Stream) Map(name string, mapper Mapper) *Stream

+ func (s *Stream) Merge(name string, streams ...*Stream) *Stream

+ func (s *Stream) Print(name string) *Stream

+ func (s *Stream) Process(name string, p Processor) *Stream

+ type StreamBuilder struct

+ func NewStreamBuilder() *StreamBuilder

+ func (sb *StreamBuilder) Build() *Topology

+ func (sb *StreamBuilder) Source(name string, source Source) *Stream

+ type Task interface

+ Close func() error

+ OnError func(fn ErrorFunc)

+ Start func() error

+ func NewTask(topology *Topology) Task

+ type Topology struct

+ func (t Topology) Processors() []Node

+ func (t Topology) Sources() map[Source]Node

+ type TopologyBuilder struct

+ func NewTopologyBuilder() *TopologyBuilder

+ func (tb *TopologyBuilder) AddProcessor(name string, processor Processor, parents []Node) Node

+ func (tb *TopologyBuilder) AddSource(name string, source Source) Node

+ func (tb *TopologyBuilder) Build() *Topology

Other modules containing this package

github.com/msales/streams/v2

github.com/msales/streams/v4

github.com/msales/streams/v6

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL