worker

Documentation

Index

• func New[QueueItem comparable](name string, clk clock.WithTicker) component.Declared[Api[QueueItem]]
• type Api
  • func (api Api[QueueItem]) Enqueue(item QueueItem)
  • func (api Api[QueueItem]) EnqueueDelayed(item QueueItem, delay time.Duration)
  • func (api Api[QueueItem]) SetBeforeStart(beforeStart BeforeStart)
  • func (api Api[QueueItem]) SetExecutor(executor Executor[QueueItem], prereqs map[string]Prereq)
• type Args
• type BeforeStart
• type Deps
• type Executor
• type Options
• type Prereq
  • func HasSyncedPrereq(hasSynced cache.InformerSynced) Prereq
  • func InformerPrereq(informer cache.SharedIndexInformer) Prereq
• type State

Functions

func New

func New[QueueItem comparable](
	name string,
	clk clock.WithTicker,
) component.Declared[Api[QueueItem]]

Types

type Api

type Api[QueueItem comparable] struct {
	// contains filtered or unexported fields
}

func (Api[QueueItem]) Enqueue

func (api Api[QueueItem]) Enqueue(item QueueItem)

func (Api[QueueItem]) EnqueueDelayed

func (api Api[QueueItem]) EnqueueDelayed(item QueueItem, delay time.Duration)

func (Api[QueueItem]) SetBeforeStart

func (api Api[QueueItem]) SetBeforeStart(beforeStart BeforeStart)

func (Api[QueueItem]) SetExecutor

func (api Api[QueueItem]) SetExecutor(executor Executor[QueueItem], prereqs map[string]Prereq)

type Args

type Args struct {
	Name  string
	Clock clock.WithTicker
}

type BeforeStart

type BeforeStart func(context.Context) (context.Context, error)

type Deps

type Deps struct {
	Observer component.Dep[observer.Observer]
}

type Executor

type Executor[QueueItem comparable] func(ctx context.Context, item QueueItem) error

type Options

type Options struct {
	WorkerCount *int
	BaseDelay   *time.Duration
	MaxDelay    *time.Duration
}

type Prereq

type Prereq interface {
	Wait(ctx context.Context) bool

	IsReady() bool
}

A prerequisite condition before a worker can starts running.

Worker starts running as long as Wait returns true the first time. IsReady may return false subsequently, but it would only affect health checks but not worker liveness.

func HasSyncedPrereq

func HasSyncedPrereq(hasSynced cache.InformerSynced) Prereq

func InformerPrereq

func InformerPrereq(informer cache.SharedIndexInformer) Prereq

type State

type State[QueueItem comparable] struct {
	// contains filtered or unexported fields
}