pipelines

Aug 6, 2025

Changes in this version

type Pipeline

+ func (p *Pipeline[T, U]) Broadcast(num int, streamFunc StreamFunc[T, U]) Pipelines[T, U]

+ func (p *Pipeline[T, U]) Copy(num int) Pipelines[T, U]

+ type Pipelines []*Pipeline[T, U]

+ func (p Pipelines[T, U]) Close()

+ func (p Pipelines[T, U]) Count() int

+ func (p Pipelines[T, U]) Get(index int) (*Pipeline[T, U], error)

+ func (p Pipelines[T, U]) Process(processor datastreams.ProcessFunc[T], params ...datastreams.Params) Pipelines[T, U]

+ func (p Pipelines[T, U]) Start(streamFunc StreamFunc[T, U]) Pipelines[T, U]

+ func (p Pipelines[T, U]) Wait()

Apr 25, 2025

Changes in this version

type Pipeline

+ func (p *Pipeline[T, U]) Expand(expander datastreams.ExpandFunc[T, U], params ...datastreams.Params) datastreams.DataStream[U]

Mar 27, 2025

Changes in this version

+ type Opts struct

+ BufferSize int

+ type Pipeline struct

+ func New[T any, U any](ctx context.Context, sourcer datastreams.Sourcer[T], errStream chan error) *Pipeline[T, U]

+ func (p *Pipeline[T, U]) Close()

+ func (p *Pipeline[T, U]) Errors() <-chan error

+ func (p *Pipeline[T, U]) In() <-chan T

+ func (p *Pipeline[T, U]) Map(mapper datastreams.TransformFunc[T, U], params ...datastreams.Params) datastreams.DataStream[U]

+ func (p *Pipeline[T, U]) Out() <-chan U

+ func (p *Pipeline[T, U]) Process(processor datastreams.ProcessFunc[T], params ...datastreams.Params) *Pipeline[T, U]

+ func (p *Pipeline[T, U]) Sink(sinker datastreams.Sinker[U]) error

+ func (p *Pipeline[T, U]) Start(streamFunc StreamFunc[T, U]) *Pipeline[T, U]

+ func (p *Pipeline[T, U]) Stream(streamFunc StreamFunc[T, U]) datastreams.DataStream[U]

+ func (p *Pipeline[T, U]) Tee(params ...datastreams.Params) (datastreams.DataStream[U], datastreams.DataStream[U])

+ func (p *Pipeline[T, U]) ToSource() datastreams.Sourcer[U]

+ func (p *Pipeline[T, U]) Wait()

+ type StreamFunc func(stream datastreams.DataStream[T]) datastreams.DataStream[U]