README
¶
tombstreams
A stream processing library for Go that combines go-streams and tombs.
Most of the flow capabilities from go-streams are supported. Go channels are the only supported connector. Tombs were added to provide a way to cancel a pipeline and surface any errors.
Documentation
¶
Index ¶
- func DoStream(outlet Outlet, inlet Inlet)
- func GenerateIDs(ctx context.Context, ids []string) <-chan interface{}
- type ChanSink
- type ChanSource
- type Filter
- type FilterFunc
- type FlatMap
- type FlatMapFunc
- type Flow
- type IgnoreSink
- type Inlet
- type Map
- type MapFunc
- type Outlet
- type PassThrough
- type Sink
- type Source
- type StdoutSink
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func GenerateIDs ¶
Types ¶
type ChanSink ¶
type ChanSink struct {
Out chan interface{}
}
ChanSink sends data to the output channel
func NewChanSink ¶
func NewChanSink(out chan interface{}) *ChanSink
NewChanSink returns a new ChanSink instance
type ChanSource ¶
type ChanSource struct {
// contains filtered or unexported fields
}
ChanSource streams data from the input channel
func NewChanSource ¶
func NewChanSource(t *tomb.Tomb, in <-chan interface{}) *ChanSource
NewChanSource returns a new ChanSource instance
func (*ChanSource) Out ¶
func (cs *ChanSource) Out() <-chan interface{}
Out returns an output channel for sending data
func (*ChanSource) Via ¶
func (cs *ChanSource) Via(_flow Flow) Flow
Via streams data through the given flow
type Filter ¶
type Filter struct {
FilterF FilterFunc
// contains filtered or unexported fields
}
Filter filters the incoming elements using a predicate. If the predicate returns true the element is passed downstream, if it returns false the element is discarded.
in -- 1 -- 2 ---- 3 -- 4 ------ 5 --
| | | | |
[---------- FilterFunc -----------]
| | |
out -- 1 -- 2 ------------------ 5 --
func NewFilter ¶
func NewFilter(t *tomb.Tomb, filterFunc FilterFunc, parallelism uint) *Filter
NewFilter returns a new Filter instance. filterFunc is the filter predicate function. parallelism is the flow parallelism factor. In case the events order matters, use parallelism = 1.
func (*Filter) In ¶
func (f *Filter) In() chan<- interface{}
In returns an input channel for receiving data
type FilterFunc ¶
FilterFunc is a filter predicate function.
type FlatMap ¶
type FlatMap struct {
FlatMapF FlatMapFunc
// contains filtered or unexported fields
}
FlatMap takes one element and produces zero, one, or more elements.
in -- 1 -- 2 ---- 3 -- 4 ------ 5 --
| | | | |
[---------- FlatMapFunc ----------]
| | | | |
out -- 1' - 2' -------- 4'- 4”- 5' -
func NewFlatMap ¶
func NewFlatMap(t *tomb.Tomb, flatMapFunc FlatMapFunc, parallelism uint) *FlatMap
NewFlatMap returns a new FlatMap instance. flatMapFunc is the FlatMap transformation function. parallelism is the flow parallelism factor. In case the events order matters, use parallelism = 1.
func (*FlatMap) In ¶
func (fm *FlatMap) In() chan<- interface{}
In returns an input channel for receiving data
type FlatMapFunc ¶
type FlatMapFunc func(interface{}) ([]interface{}, error)
FlatMapFunc is a FlatMap transformation function.
type Flow ¶
Flow is a set of stream processing steps that has one open input and one open output.
type IgnoreSink ¶
type IgnoreSink struct {
// contains filtered or unexported fields
}
IgnoreSink sends items to /dev/null
func NewIgnoreSink ¶
func NewIgnoreSink(t *tomb.Tomb) *IgnoreSink
NewIgnoreSink returns a new IgnoreSink instance
func (*IgnoreSink) In ¶
func (ignore *IgnoreSink) In() chan<- interface{}
In returns an input channel for receiving data
type Inlet ¶
type Inlet interface {
In() chan<- interface{}
}
Inlet is a type that exposes one open input. Implemented by the Flow and Sink.
type Map ¶
type Map struct {
MapF MapFunc
// contains filtered or unexported fields
}
Map takes one element and produces one element.
in -- 1 -- 2 ---- 3 -- 4 ------ 5 --
| | | | |
[----------- MapFunc -------------]
| | | | |
out -- 1' - 2' --- 3' - 4' ----- 5' -
func NewMap ¶
NewMap returns a new Map instance. mapFunc is the Map transformation function. parallelism is the flow parallelism factor. In case the events order matters, use parallelism = 1.
func (*Map) In ¶
func (m *Map) In() chan<- interface{}
In returns an input channel for receiving data
type MapFunc ¶
type MapFunc func(interface{}) (interface{}, error)
MapFunc is a Map transformation function.
type Outlet ¶
type Outlet interface {
Out() <-chan interface{}
Tomb() *tomb.Tomb
}
Outlet is a type that exposes one open output. Implemented by the Source and Flow.
type PassThrough ¶
type PassThrough struct {
// contains filtered or unexported fields
}
PassThrough produces the received element as is.
in -- 1 -- 2 ---- 3 -- 4 ------ 5 --
| | | | |
out -- 1 -- 2 ---- 3 -- 4 ------ 5 --
func NewPassThrough ¶
func NewPassThrough(t *tomb.Tomb) *PassThrough
NewPassThrough returns a new PassThrough instance.
func (*PassThrough) In ¶
func (pt *PassThrough) In() chan<- interface{}
In returns an input channel for receiving data
func (*PassThrough) Out ¶
func (pt *PassThrough) Out() <-chan interface{}
Out returns an output channel for sending data
func (*PassThrough) Tomb ¶
func (pt *PassThrough) Tomb() *tomb.Tomb
func (*PassThrough) Via ¶
func (pt *PassThrough) Via(flow Flow) Flow
Via streams data through the given flow
type Sink ¶
type Sink interface {
Inlet
}
Sink is a set of stream processing steps that has one open input. Can be used as a Subscriber.
type StdoutSink ¶
type StdoutSink struct {
// contains filtered or unexported fields
}
StdoutSink sends items to stdout
func NewStdoutSink ¶
func NewStdoutSink(t *tomb.Tomb) *StdoutSink
NewStdoutSink returns a new StdoutSink instance
func (*StdoutSink) In ¶
func (stdout *StdoutSink) In() chan<- interface{}
In returns an input channel for receiving data
Source Files