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.1

Oct 9, 2018

v1.0.0

Aug 31, 2018

Changes in this version

+ type NodeMessage struct

+ Msg *Message

+ Node Node

type Pipe

+ Queue func() []NodeMessage

type ProcessorPipe

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

v0

v0.5.4

Aug 3, 2018

v0.5.3

Aug 3, 2018

v0.5.2

Jul 5, 2018

v0.5.1

Jul 5, 2018

v0.5.0

Jun 27, 2018

Changes in this version

type Message

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

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

v0.4.1

Jun 21, 2018

v0.4.0

Jun 21, 2018

Changes in this version

+ 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 PassThroughProcessor struct

+ func (p *PassThroughProcessor) Close() error

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

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

type Processor

+ func NewFlatMapProcessor(fn FlatMapper) Processor

+ func NewPassThroughProcessor() Processor

type Stream

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

v0.3.0

Jun 13, 2018

Changes in this version

type BranchProcessor

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

type FilterProcessor

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

type MapProcessor

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

type MergeProcessor

+ 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) Empty() bool

type Node

+ Name func() string

+ WithPipe func(Pipe)

+ type Pipe interface

+ Commit func() error

+ Forward func(*Message) error

+ ForwardToChild func(*Message, int) error

type PrintProcessor

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

type Processor

+ WithPipe func(Pipe)

type ProcessorNode

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

+ func (n *ProcessorNode) Name() string

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

+ type ProcessorPipe struct

+ func NewProcessorPipe(t Task) *ProcessorPipe

+ func (p *ProcessorPipe) Commit() error

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

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

+ func (p *ProcessorPipe) SetNode(n Node)

type SourceNode

+ func NewSourceNode(name string) *SourceNode

+ func (n *SourceNode) Name() string

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

v0.2.0

Jun 8, 2018

v0.1.1

Jun 7, 2018

v0.1.0

May 7, 2018

Changes in this version

+ type BranchProcessor struct

+ func (p *BranchProcessor) Close() error

+ func (p *BranchProcessor) Process(key, value interface{}) error

+ func (p *BranchProcessor) WithContext(ctx Context)

+ type Context interface

+ Commit func() error

+ Forward func(key, value interface{}) error

+ ForwardToChild func(key, value interface{}, child int) error

+ Logger func() log.Logger

+ Stats func() stats.Stats

+ type ErrorFunc func(error)

+ type FilterProcessor struct

+ func (p *FilterProcessor) Close() error

+ func (p *FilterProcessor) Process(key, value interface{}) error

+ func (p *FilterProcessor) WithContext(ctx Context)

+ type MapProcessor struct

+ func (p *MapProcessor) Close() error

+ func (p *MapProcessor) Process(key, value interface{}) error

+ func (p *MapProcessor) WithContext(ctx Context)

+ type Mapper func(key, value interface{}) (interface{}, interface{}, error)

+ type MergeProcessor struct

+ func (p *MergeProcessor) Close() error

+ func (p *MergeProcessor) Process(key, value interface{}) error

+ func (p *MergeProcessor) WithContext(ctx Context)

+ type Node interface

+ AddChild func(n Node)

+ Children func() []Node

+ Close func() error

+ Process func(key, value interface{}) error

+ WithContext func(ctx Context)

+ type Predicate func(k, v interface{}) (bool, error)

+ type PrintProcessor struct

+ func (p *PrintProcessor) Close() error

+ func (p *PrintProcessor) Process(key, value interface{}) error

+ func (p *PrintProcessor) WithContext(ctx Context)

+ type Processor interface

+ Close func() error

+ Process func(key, value interface{}) error

+ WithContext func(ctx Context)

+ func NewBranchProcessor(fns []Predicate) Processor

+ func NewFilterProcessor(fn Predicate) Processor

+ func NewMapProcessor(fn Mapper) Processor

+ func NewMergeProcessor() Processor

+ func NewPrintProcessor() Processor

+ type ProcessorContext struct

+ func NewProcessorContext(t Task, l log.Logger, s stats.Stats) *ProcessorContext

+ func (c *ProcessorContext) Commit() error

+ func (c *ProcessorContext) Forward(key, value interface{}) error

+ func (c *ProcessorContext) ForwardToChild(key, value interface{}, index int) error

+ func (c *ProcessorContext) Logger() log.Logger

+ func (c *ProcessorContext) Stats() stats.Stats

+ type ProcessorNode struct

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

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

+ func (n *ProcessorNode) Close() error

+ func (n *ProcessorNode) Process(key, value interface{}) error

+ func (n *ProcessorNode) WithContext(ctx Context)

+ type Source interface

+ Close func() error

+ Commit func() error

+ Consume func() (key, value interface{}, err error)

+ type SourceNode struct

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

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

+ func (n *SourceNode) Close() error

+ func (n *SourceNode) Process(key, value interface{}) error

+ func (n *SourceNode) WithContext(ctx Context)

+ type Stream struct

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

+ func (s *Stream) Filter(name string, pred Predicate) *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

+ Commit func() error

+ OnError func(fn ErrorFunc)

+ Start func()

+ func NewTask(topology *Topology, opts ...TaskFunc) Task

+ type TaskFunc func(*streamTask)

+ func WithLogger(logger log.Logger) TaskFunc

+ func WithStats(stats stats.Stats) TaskFunc

+ 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

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