conveyor

package module

Go to main page

Versions in this module

v0

v0.18.0

Oct 22, 2026

Changes in this version

type Conveyor

+ Defer func(ctx context.Context, fn func(err error))

v0.17.0

Oct 3, 2026

Changes in this version

type Conveyor

+ func New(options ...Option) Conveyor

+ ItemLimit func() int

+ SetItemLimit func(n int) Conveyor

+ type NodeOption interface

+ func WithName(name string) NodeOption

type Option

+ func WithDrainContext(f DrainContextFunc) Option

+ func WithDrainTimeout(d time.Duration) Option

v0.16.0

Oct 3, 2026

Changes in this version

+ func ShutdownCause(pre context.Context, err error) error

+ func UntilShutdown(ctx context.Context) context.Context

+ type DrainContextFunc func(cause error) (context.Context, context.CancelFunc)

type Option

+ func OptDrainContextFunc(f DrainContextFunc) Option

+ func OptDrainTimeout(d time.Duration) Option

+ type TaskError interface

+ Unit func() Unit

+ Unwrap func() error

+ type TaskGroup interface

+ Finished func() <-chan struct{}

+ Wait func(ctx context.Context) error

v0.15.0

Sep 30, 2026

Changes in this version

+ type ItemError interface

+ Unit func() Unit

+ type RunError interface

+ DrainError func() error

+ ItemErrors func() []error

+ Unwrap func() error

v0.14.0

Sep 29, 2026

Changes in this version

type Conveyor

+ NoAbortPoint func() Unit

+ SetNoAbortPoint func(node Unit) Conveyor

type RetainableStage

+ RetainFor func(ctx context.Context, bgOp func() error) Wave

v0.13.0

Sep 24, 2026

Changes in this version

type Conveyor

+ StartingStage func() RetainableStage

+ type RetainableStage interface

+ Retain func(ctx context.Context, bgOp func() error) Wave

v0.12.0

Sep 16, 2026

Changes in this version

type FanOut

+ Backpressure func() FanOutBackpressure

+ Retain func(ctx context.Context) Wave

+ SetBackpressure func(mode FanOutBackpressure) FanOut

+ type FanOutBackpressure int

+ const BackpressureBalanced

+ const BackpressureBuffered

+ const BackpressureStrict

+ func (m FanOutBackpressure) String() string

v0.11.1

Sep 11, 2026

v0.11.0

Sep 11, 2026

Changes in this version

type FanOut

+ Admission func() FanOutAdmission

+ SetAdmission func(a FanOutAdmission) FanOut

+ type FanOutAdmission int

+ const AdmitByLimit

+ const AdmitByPools

+ const AdmitByPoolsStrict

+ func (a FanOutAdmission) String() string

v0.10.0

Sep 10, 2026

Changes in this version

type FanOut

+ Schedule func(ctx context.Context, tasks ...Task) error

+ Wait func(ctx context.Context) error

type Wave

+ Wait func(ctx context.Context) error

v0.9.2

Aug 10, 2026

v0.9.1

Aug 6, 2026

Changes in this version

type Conveyor

+ func (c *Conveyor) ItemsLimit() int

+ func (c *Conveyor) SetItemsLimit(n int) *Conveyor

v0.9.0

Aug 5, 2026

Changes in this version

+ var ErrConveyorAlreadyRunning = errors.New("conveyor is already running")

+ var ErrForeignContext = fmt.Errorf("%w: not derived from a conveyor item's ctx", ErrInvalidContext)

+ var ErrInvalidContext = errors.New("invalid context")

+ var ErrStaleContext = fmt.Errorf("%w: item is finished", ErrInvalidContext)

+ func ItemNoFromContext(ctx context.Context) (no int64, ok bool)

+ type AnyUnitOption interface

+ type Branch interface

+ NewTask func(fn TaskFunc) Task

+ NewTasks func(count int, fn func(ctx context.Context, index int) error) Task

+ NewTasksChan func(ch <-chan TaskFunc) Task

+ NewTasksGen func(gen iter.Seq[TaskFunc]) Task

+ type Conveyor struct

+ func NewConveyor(options ...Option) *Conveyor

+ func (c *Conveyor) DebugUnitOccupants() []UnitOccupants

+ func (c *Conveyor) Run(ctx context.Context, itemProcessor ItemProcessor) error

+ func (c *Conveyor) StartUnit() Unit

+ func (c *Conveyor) Stats() Stats

+ func (s Conveyor) AddFanOut(opts ...AnyUnitOption) FanOut

+ func (s Conveyor) AddStage(opts ...AnyUnitOption) Stage

+ type FanOut interface

+ AddLane func(opts ...AnyUnitOption) Lane

+ AddPool func(opts ...AnyUnitOption) Pool

+ Branches func() []Branch

+ Detach func(ctx context.Context) Wave

+ Limit func() int

+ MoveTo func(ctx context.Context, tasks Tasks, joins ...Wave) error

+ QueueSize func() int

+ SetLimit func(limit int) FanOut

+ SetQueueSize func(size int) FanOut

+ TryMoveTo func(ctx context.Context, tasks Tasks, joins ...Wave) (entered bool, err error)

+ type Gauge struct

+ Last int

+ Max int

+ Min int

+ type ItemProcessor func(ctx context.Context) error

+ type Lane interface

+ AddFanOut func(opts ...AnyUnitOption) FanOut

+ AddStage func(opts ...AnyUnitOption) Stage

+ NewTask func(fn TaskFunc) Task

+ NewTasks func(count int, fn func(ctx context.Context, index int) error) Task

+ NewTasksChan func(ch <-chan TaskFunc) Task

+ NewTasksGen func(gen iter.Seq[TaskFunc]) Task

+ type OptName string

+ type Option func(c *Conveyor)

+ func OptShutdownContext(factory ShutdownContextFactory) Option

+ type Pool interface

+ Limit func() int

+ NewTask func(fn TaskFunc) Task

+ NewTasks func(count int, fn func(ctx context.Context, index int) error) Task

+ NewTasksChan func(ch <-chan TaskFunc) Task

+ NewTasksGen func(gen iter.Seq[TaskFunc]) Task

+ SetLimit func(limit int) Pool

+ type ShutdownContextFactory func(cause error) (context.Context, context.CancelFunc)

+ type ShutdownError interface

+ Cause func() error

+ type Stage interface

+ Limit func() int

+ MoveTo func(ctx context.Context, joins ...Wave) error

+ QueueSize func() int

+ Retain func(ctx context.Context, bgOp func() error) Wave

+ SetLimit func(limit int) Stage

+ SetQueueSize func(size int) Stage

+ TryMoveTo func(ctx context.Context, joins ...Wave) (entered bool, err error)

+ type Stats struct

+ InFlight Gauge

+ LiveWorkers Gauge

+ Units []UnitStat

+ type Task struct

+ type TaskFunc = func(ctx context.Context) error

+ type Tasks []Task

+ func (ts *Tasks) Add(tasks ...Task) *Tasks

+ type Unit interface

+ type UnitOccupants struct

+ InBody []int64

+ InQueue []int64

+ Unit Unit

+ type UnitStat struct

+ Limit int

+ Occupied Gauge

+ Queued Gauge

+ Unit Unit

+ type Wave interface

+ Err func() error

+ Finished func() <-chan struct{}

+ Started func() <-chan struct{}

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