Versions in this module Expand all Collapse all v0 v0.3.0 Oct 10, 2026 Changes in this version + type Backpressure int + const BackpressureBlock + const BackpressureDrop + func (b Backpressure) String() string + type Collector struct + func (c *Collector[ITEM]) Engine() *Engine[ITEM] + func (c *Collector[ITEM]) Push(item ITEM) error + func (c *Collector[ITEM]) PushAll(items ...ITEM) error + func (c *Collector[ITEM]) TryPush(item ITEM) error + type Config struct + Backpressure Backpressure + BatchSize int + BufferCap int + Concurrency int + DeferOnThreshold bool + Handler basic.Handler[ITEM] + Interval time.Duration + Logf func(format string, args ...any) + MaxPending int + Mode Mode + Name string + OnDrop func(basic.DropEvent) + OnExecute func(basic.ExecuteEvent) + PeriodicHandler basic.Handler[ITEM] + QueueSize int + Recover basic.RecoverFunc + RecoverKeep int + SizeHandler basic.Handler[ITEM] + type Engine struct + func New[ITEM any](cfg Config[ITEM], opts ...Option[ITEM]) (*Engine[ITEM], error) + func (e *Engine[ITEM]) Await() basic.Stats + func (e *Engine[ITEM]) Close() + func (e *Engine[ITEM]) Collect(item ITEM) error + func (e *Engine[ITEM]) Collector() *Collector[ITEM] + func (e *Engine[ITEM]) Configure(opts ...Option[ITEM]) + func (e *Engine[ITEM]) Done() <-chan struct{} + func (e *Engine[ITEM]) Flush() error + func (e *Engine[ITEM]) FlushAndDrain() error + func (e *Engine[ITEM]) Pause() + func (e *Engine[ITEM]) Pending() int + func (e *Engine[ITEM]) Resume() + func (e *Engine[ITEM]) Start() + func (e *Engine[ITEM]) Stats() basic.Stats + func (e *Engine[ITEM]) Status() basic.Status + func (e *Engine[ITEM]) TryCollect(item ITEM) error + func (e *Engine[ITEM]) Wait(ctx context.Context) error + type Mode int + const ModeCompensate + const ModeConcurrent + const ModeInterval + const ModeThreshold + const ModeTrigger + func (m Mode) String() string + type Option func(*Options[ITEM]) + func WithBackpressure[ITEM any](bp Backpressure) Option[ITEM] + func WithBatchSize[ITEM any](size int) Option[ITEM] + func WithConcurrency[ITEM any](n int) Option[ITEM] + func WithDeferOnThreshold[ITEM any](v bool) Option[ITEM] + func WithInterval[ITEM any](d time.Duration) Option[ITEM] + func WithLogf[ITEM any](fn func(format string, args ...any)) Option[ITEM] + func WithMaxPending[ITEM any](n int) Option[ITEM] + func WithOnDrop[ITEM any](fn func(basic.DropEvent)) Option[ITEM] + func WithOnExecute[ITEM any](fn func(basic.ExecuteEvent)) Option[ITEM] + func WithPeriodicHandler[ITEM any](h basic.Handler[ITEM]) Option[ITEM] + func WithRecover[ITEM any](r basic.RecoverFunc) Option[ITEM] + func WithSizeHandler[ITEM any](h basic.Handler[ITEM]) Option[ITEM] + type Options struct + Backpressure Backpressure + BatchSize int + Concurrency int + DeferOnThreshold bool + Interval time.Duration + Logf func(format string, args ...any) + MaxPending int + OnDrop func(basic.DropEvent) + OnExecute func(basic.ExecuteEvent) + PeriodicHandler basic.Handler[ITEM] + Recover basic.RecoverFunc + SetBackpressure bool + SetConcurrency bool + SetDeferOnThreshold bool + SetInterval bool + SetMaxPending bool + SetSize bool + SizeHandler basic.Handler[ITEM] + type Trigger uint8 + const TriggerArrival + const TriggerPeriodic + const TriggerThreshold + func (t Trigger) String() string