pipeline

package module

Go to main page

Versions in this module

v3

v3.2.0

Feb 18, 2026

Changes in this version

+ type Composer struct

+ func (c Composer[In, Out]) Context() Context

+ func (c Composer[In, Out]) Inputs() MultiChannelReceiver[In]

+ func (c Composer[In, Out]) Outputs() MultiChannelSender[Out]

+ type Config struct

+ Composer func(Composer[In, Out])

+ InputBufferSize int

+ InputChannels int

+ Name string

+ OutputBufferSize int

+ OutputChannels int

+ type InputChannel interface

type MultiChannelReceiver

+ func NewMultiChannelReceiver[T any, C InputChannel[T]](in ...C) MultiChannelReceiver[T]

+ func (m MultiChannelReceiver[T]) SinkAt(ctx context.Context, index int) []T

+ func (m MultiChannelReceiver[T]) SinkAtIter(ctx context.Context, index int) iter.Seq[T]

type MultiChannelSender

+ func NewMultiChannelSender[T any, C OutputChannel[T]](out ...C) MultiChannelSender[T]

+ func (m MultiChannelSender[T]) Link(ctx Context, index int, in <-chan T) error

+ func (m MultiChannelSender[T]) LinkAll(ctx Context, in MultiChannelReceiver[T]) error

+ type OutputChannel interface

v3.1.0

Feb 17, 2026

Changes in this version

+ type Context struct

+ func (c Context) Context() context.Context

+ func (c Context) Go(name string, fn func(ctx context.Context) error)

type MultiChannelSender

+ func (m MultiChannelSender[T]) Receiver(index int) <-chan T

+ func (m MultiChannelSender[T]) Receivers() MultiChannelReceiver[T]

+ func (m MultiChannelSender[T]) SendRoundRobin(ctx context.Context, values ...T) error

+ func (m MultiChannelSender[T]) SendToAll(ctx context.Context, values ...T) error

v3.0.0

Feb 17, 2026

Changes in this version

+ const StateCreated

+ const StateDone

+ const StateInvalid

+ const StateStarted

+ const StateWaiting

+ func Aggregate[T any](name string, pipe *Pipe, in <-chan T, out chan<- []T)

+ func Batch[In any, Out any](name string, pipe *Pipe, batcher func(context.Context, []In) (*Out, error), ...)

+ func ExpandSlice[T any](name string, pipe *Pipe, in <-chan []T, out chan<- T)

+ func Expand[In any, Out any](name string, pipe *Pipe, ...)

+ func FanIn[T any](name string, pipe *Pipe, out chan<- T, in ...<-chan T)

+ func FanOutRoundRobin[T any](name string, pipe *Pipe, in <-chan T, out ...chan<- T)

+ func FanOut[T any](name string, pipe *Pipe, in <-chan T, out ...chan<- T)

+ func Filter[T any](name string, pipe *Pipe, filter func(context.Context, T) (bool, error), ...)

+ func Flatten[T any](name string, pipe *Pipe, in <-chan []T, out chan<- T)

+ func Limit[T any](name string, pipe *Pipe, n int, in <-chan T, out chan<- T)

+ func ParallelTransform[In any, Out any](name string, pipe *Pipe, workers int, ...)

+ func Reduce[T any, Acc any](name string, pipe *Pipe, initial Acc, ...)

+ func Split[T any](name string, pipe *Pipe, selector func(context.Context, T) int, in <-chan T, ...)

+ func Transform[In any, Out any](name string, pipe *Pipe, transformer func(context.Context, In) (*Out, error), ...)

+ type MultiChannelReceiver []chan T

+ func (m MultiChannelReceiver[T]) At(index int) <-chan T

+ func (m MultiChannelReceiver[T]) Iter() iter.Seq[<-chan T]

+ func (m MultiChannelReceiver[T]) Len() int

+ type MultiChannelSender []chan T

+ func (m MultiChannelSender[T]) At(index int) chan<- T

+ func (m MultiChannelSender[T]) Iter() iter.Seq[chan<- T]

+ func (m MultiChannelSender[T]) Len() int

+ func (m MultiChannelSender[T]) Send(ctx context.Context, index int, values ...T) error

+ type Pipe struct

+ func (p Pipe) Context() context.Context

+ type Pipeline struct

+ func Chain[In any, Middle any, Out any](prev *Pipeline[In, Middle], configs ...PipelineConfig[Middle, Out]) ([]*Pipeline[Middle, Out], error)

+ func NewPipeline[In any, Out any](ctx context.Context, cfg PipelineConfig[In, Out]) (*Pipeline[In, Out], context.Context)

+ func (p *Pipeline[In, Out]) CloseAllInputs()

+ func (p *Pipeline[In, Out]) Context() context.Context

+ func (p *Pipeline[In, Out]) Inputs() MultiChannelSender[In]

+ func (p *Pipeline[In, Out]) Outputs() MultiChannelReceiver[Out]

+ func (p *Pipeline[In, Out]) Start() error

+ func (p *Pipeline[In, Out]) State() *PipelineState

+ func (p *Pipeline[In, Out]) Wait() error

+ type PipelineConfig struct

+ Executor PipelineExecutor[In, Out]

+ InputBufferSize int

+ InputChannels int

+ Name string

+ OutputBufferSize int

+ OutputChannels int

+ StartImmediately bool

+ type PipelineExecutor func(*PipelineParameters[In, Out])

+ type PipelineGroup struct

+ func NewPipelineGroup[In any, Out any]() *PipelineGroup[In, Out]

+ func (g *PipelineGroup[In, Out]) Add(pipelines ...*Pipeline[In, Out])

+ func (g *PipelineGroup[In, Out]) Start() error

+ func (g *PipelineGroup[In, Out]) Wait(ctx context.Context) error

+ type PipelineParameters struct

+ In MultiChannelReceiver[In]

+ Out MultiChannelSender[Out]

+ Pipe *Pipe

+ type PipelineState struct

+ func (s *PipelineState) Is(state int32) bool

+ func (s *PipelineState) String() string

Other modules containing this package

github.com/schraf/pipeline

github.com/schraf/pipeline/v2

github.com/schraf/pipeline/v4

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