toc

Mar 21, 2026

Changes in this version

type Options

+ MaxWIP int

type Stage

+ func (s *Stage[T, R]) MaxWIP() int

+ func (s *Stage[T, R]) SetMaxWIP(n int) int

type Stats

+ Admitted int64

+ MaxWIP int

+ RopeWaitCount int64

+ RopeWaitNs int64

Mar 19, 2026

Changes in this version

+ func FromChan[T any](ch <-chan T) <-chan rslt.Result[T]

Mar 19, 2026

Changes in this version

+ type Join struct

+ func NewJoin[A, B, R any](ctx context.Context, srcA <-chan rslt.Result[A], srcB <-chan rslt.Result[B], ...) *Join[R]

+ func (j *Join[R]) Out() <-chan rslt.Result[R]

+ func (j *Join[R]) Stats() JoinStats

+ func (j *Join[R]) Wait() error

+ type JoinStats struct

+ Combined int64

+ DiscardedA int64

+ DiscardedB int64

+ Errors int64

+ ExtraA int64

+ ExtraB int64

+ OutputBlockedTime time.Duration

+ ReceivedA int64

+ ReceivedB int64

+ type Merge struct

+ func NewMerge[T any](ctx context.Context, sources ...<-chan rslt.Result[T]) *Merge[T]

+ func (m *Merge[T]) Out() <-chan rslt.Result[T]

+ func (m *Merge[T]) Stats() MergeStats

+ func (m *Merge[T]) Wait() error

+ type MergeStats struct

+ Dropped int64

+ Forwarded int64

+ Received int64

+ SourceDropped []int64

+ SourceForwarded []int64

+ SourceReceived []int64

+ type MissingResultError struct

+ Source string

+ func (e *MissingResultError) Error() string

Mar 18, 2026

Changes in this version

type Options

+ TrackAllocations bool

type Stats

+ AllocTrackingActive bool

+ ObservedAllocBytes uint64

+ ObservedAllocObjects uint64

+ type Tee struct

+ func NewTee[T any](ctx context.Context, src <-chan rslt.Result[T], n int) *Tee[T]

+ func (t *Tee[T]) Branch(i int) <-chan rslt.Result[T]

+ func (t *Tee[T]) Stats() TeeStats

+ func (t *Tee[T]) Wait() error

+ type TeeStats struct

+ BranchBlockedTime []time.Duration

+ BranchDelivered []int64

+ FullyDelivered int64

+ PartiallyDelivered int64

+ Received int64

+ Undelivered int64

Mar 17, 2026

Changes in this version

+ var ErrClosed = errors.New("toc: stage closed")

+ type Batcher struct

+ func NewBatcher[T any](ctx context.Context, src <-chan rslt.Result[T], n int) *Batcher[T]

+ func (b *Batcher[T]) Out() <-chan rslt.Result[[]T]

+ func (b *Batcher[T]) Stats() BatcherStats

+ func (b *Batcher[T]) Wait() error

+ type BatcherStats struct

+ BatchCount int64

+ BufferedDepth int64

+ Dropped int64

+ Emitted int64

+ Forwarded int64

+ OutputBlockedTime time.Duration

+ Received int64

+ type Options struct

+ Capacity int

+ ContinueOnError bool

+ Weight func(T) int64

+ Workers int

+ type Stage struct

+ func Pipe[T, R any](ctx context.Context, src <-chan rslt.Result[T], ...) *Stage[T, R]

+ func Start[T, R any](ctx context.Context, fn func(context.Context, T) (R, error), opts Options[T]) *Stage[T, R]

+ func (s *Stage[T, R]) Cause() error

+ func (s *Stage[T, R]) CloseInput()

+ func (s *Stage[T, R]) DiscardAndCause() error

+ func (s *Stage[T, R]) DiscardAndWait() error

+ func (s *Stage[T, R]) Out() <-chan rslt.Result[R]

+ func (s *Stage[T, R]) Stats() Stats

+ func (s *Stage[T, R]) Submit(ctx context.Context, item T) error

+ func (s *Stage[T, R]) Wait() error

+ type Stats struct

+ BufferedDepth int64

+ Canceled int64

+ Completed int64

+ Dropped int64

+ Failed int64

+ Forwarded int64

+ IdleTime time.Duration

+ InFlightWeight int64

+ OutputBlockedTime time.Duration

+ Panicked int64

+ QueueCapacity int

+ Received int64

+ ServiceTime time.Duration

+ Submitted int64

+ type WeightedBatcher struct

+ func NewWeightedBatcher[T any](ctx context.Context, src <-chan rslt.Result[T], threshold int, ...) *WeightedBatcher[T]

+ func (b *WeightedBatcher[T]) Out() <-chan rslt.Result[[]T]

+ func (b *WeightedBatcher[T]) Stats() WeightedBatcherStats

+ func (b *WeightedBatcher[T]) Wait() error

+ type WeightedBatcherStats struct

+ BatchCount int64

+ BufferedDepth int64

+ BufferedWeight int64

+ Dropped int64

+ Emitted int64

+ Forwarded int64

+ OutputBlockedTime time.Duration

+ Received int64