pipeline

Dec 16, 2016

Changes in this version

+ type ChanOutput chan interface

+ func (oc ChanOutput) OutCh() <-chan interface{}

+ func (oc ChanOutput) Send(v interface{}) (err error)

+ func (oc ChanOutput) SendEndMark(s string) error

+ func NilOutput(ctx context.Context) ChanOutput

+ type Output interface

+ Send func(interface{}) error

Aug 19, 2016

Changes in this version

+ func IsEndMark(err error) bool

+ type Acceptor interface

+ Accept func(context.Context, chan interface{}, OutputChannel)

+ type Destination interface

+ Done func() <-chan struct{}

+ Reset func()

+ type EndMark struct

+ func (e EndMark) EndMark() bool

+ func (e EndMark) Error() string

+ type EndMarker interface

+ EndMark func() bool

+ type OutputChannel chan interface

+ func (oc OutputChannel) OutCh() <-chan interface{}

+ func (oc OutputChannel) Send(v interface{}) (err error)

+ func (oc OutputChannel) SendEndMark(s string) error

+ type Pipeline struct

+ func New() *Pipeline

+ func (p *Pipeline) Add(n Acceptor)

+ func (p *Pipeline) Done() <-chan struct{}

+ func (p *Pipeline) Run(ctx context.Context) (err error)

+ func (p *Pipeline) SetDestination(d Destination)

+ func (p *Pipeline) SetSource(s Source)

+ type Source interface

+ Reset func()

+ Start func(context.Context, OutputChannel)