workqueue

May 22, 2023

Changes in this version

type DelayingInterface

+ func NewDelayingQueueWithConfig(config DelayingQueueConfig) DelayingInterface

+ type DelayingQueueConfig struct

+ Clock clock.WithTicker

+ MetricsProvider MetricsProvider

+ Name string

+ Queue Interface

+ type QueueConfig struct

+ Clock clock.WithTicker

+ MetricsProvider MetricsProvider

+ Name string

type RateLimitingInterface

+ func NewRateLimitingQueueWithConfig(rateLimiter RateLimiter, config RateLimitingQueueConfig) RateLimitingInterface

+ func NewRateLimitingQueueWithDelayingInterface(di DelayingInterface, rateLimiter RateLimiter) RateLimitingInterface

+ type RateLimitingQueueConfig struct

+ Clock clock.WithTicker

+ DelayingQueue DelayingInterface

+ MetricsProvider MetricsProvider

+ Name string

type Type

+ func NewWithConfig(config QueueConfig) *Type

May 16, 2023

Changes in this version

+ func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc, ...)

+ func SetProvider(metricsProvider MetricsProvider)

+ func WithChunkSize(c int) func(*options)

+ type BucketRateLimiter struct

+ func (r *BucketRateLimiter) Forget(item interface{})

+ func (r *BucketRateLimiter) NumRequeues(item interface{}) int

+ func (r *BucketRateLimiter) When(item interface{}) time.Duration

+ type CounterMetric interface

+ Inc func()

+ type DelayingInterface interface

+ AddAfter func(item interface{}, duration time.Duration)

+ func NewDelayingQueue() DelayingInterface

+ func NewDelayingQueueWithCustomClock(clock clock.WithTicker, name string) DelayingInterface

+ func NewDelayingQueueWithCustomQueue(q Interface, name string) DelayingInterface

+ func NewNamedDelayingQueue(name string) DelayingInterface

+ type DoWorkPieceFunc func(piece int)

+ type GaugeMetric interface

+ Dec func()

+ Inc func()

+ type HistogramMetric interface

+ Observe func(float64)

+ type Interface interface

+ Add func(item interface{})

+ Done func(item interface{})

+ Get func() (item interface{}, shutdown bool)

+ Len func() int

+ ShutDown func()

+ ShutDownWithDrain func()

+ ShuttingDown func() bool

+ type ItemExponentialFailureRateLimiter struct

+ func (r *ItemExponentialFailureRateLimiter) Forget(item interface{})

+ func (r *ItemExponentialFailureRateLimiter) NumRequeues(item interface{}) int

+ func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration

+ type ItemFastSlowRateLimiter struct

+ func (r *ItemFastSlowRateLimiter) Forget(item interface{})

+ func (r *ItemFastSlowRateLimiter) NumRequeues(item interface{}) int

+ func (r *ItemFastSlowRateLimiter) When(item interface{}) time.Duration

+ type MaxOfRateLimiter struct

+ func (r *MaxOfRateLimiter) Forget(item interface{})

+ func (r *MaxOfRateLimiter) NumRequeues(item interface{}) int

+ func (r *MaxOfRateLimiter) When(item interface{}) time.Duration

+ type MetricsProvider interface

+ NewAddsMetric func(name string) CounterMetric

+ NewDepthMetric func(name string) GaugeMetric

+ NewLatencyMetric func(name string) HistogramMetric

+ NewLongestRunningProcessorSecondsMetric func(name string) SettableGaugeMetric

+ NewRetriesMetric func(name string) CounterMetric

+ NewUnfinishedWorkSecondsMetric func(name string) SettableGaugeMetric

+ NewWorkDurationMetric func(name string) HistogramMetric

+ type Options func(*options)

+ type RateLimiter interface

+ Forget func(item interface{})

+ NumRequeues func(item interface{}) int

+ When func(item interface{}) time.Duration

+ func DefaultControllerRateLimiter() RateLimiter

+ func DefaultItemBasedRateLimiter() RateLimiter

+ func NewItemExponentialFailureRateLimiter(baseDelay time.Duration, maxDelay time.Duration) RateLimiter

+ func NewItemFastSlowRateLimiter(fastDelay, slowDelay time.Duration, maxFastAttempts int) RateLimiter

+ func NewMaxOfRateLimiter(limiters ...RateLimiter) RateLimiter

+ func NewWithMaxWaitRateLimiter(limiter RateLimiter, maxDelay time.Duration) RateLimiter

+ type RateLimitingInterface interface

+ AddRateLimited func(item interface{})

+ Forget func(item interface{})

+ NumRequeues func(item interface{}) int

+ func NewNamedRateLimitingQueue(rateLimiter RateLimiter, name string) RateLimitingInterface

+ func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface

+ type SettableGaugeMetric interface

+ Set func(float64)

+ type SummaryMetric interface

+ Observe func(float64)

+ type Type struct

+ func New() *Type

+ func NewNamed(name string) *Type

+ func (q *Type) Add(item interface{})

+ func (q *Type) Done(item interface{})

+ func (q *Type) Get() (item interface{}, shutdown bool)

+ func (q *Type) Len() int

+ func (q *Type) ShutDown()

+ func (q *Type) ShutDownWithDrain()

+ func (q *Type) ShuttingDown() bool

+ type WithMaxWaitRateLimiter struct

+ func (w WithMaxWaitRateLimiter) Forget(item interface{})

+ func (w WithMaxWaitRateLimiter) NumRequeues(item interface{}) int

+ func (w WithMaxWaitRateLimiter) When(item interface{}) time.Duration