workqueue

Documentation

Index

• Constants
• type Config
• type IndexedWorkQueue
  • func NewIndexedWorkQueue(name string, processor Processor, cfg Config, metricsScope promutils.Scope) (IndexedWorkQueue, error)
• type Processor
• type WorkItem
• type WorkItemID
• type WorkItemInfo
• type WorkStatus
  • func WorkStatusString(s string) (WorkStatus, error)
  • func WorkStatusValues() []WorkStatus
  • func (i WorkStatus) IsAWorkStatus() bool
  • func (w WorkStatus) IsTerminal() bool
  • func (i WorkStatus) String() string

Constants

const (
	ErrNotYetStarted errors.ErrorCode = "NOT_STARTED"
)

Types

type Config

type Config struct {
	Workers            int `json:"workers" pflag:",Number of concurrent workers to start processing the queue."`
	MaxRetries         int `json:"maxRetries" pflag:",Maximum number of retries per item."`
	IndexCacheMaxItems int `json:"maxItems" pflag:",Maximum number of entries to keep in the index."`
}

Config for the queue

type IndexedWorkQueue

type IndexedWorkQueue interface {
	// Queues the item to be processed. If the item is already in the cache or has been processed before (and is still
	// in-memory), it'll not be added again.
	Queue(ctx context.Context, id WorkItemID, once WorkItem) error

	// Retrieves an item by id.
	Get(id WorkItemID) (info WorkItemInfo, found bool, err error)

	// Start must be called before queuing items into the queue.
	Start(ctx context.Context) error
}

Represents the indexed queue semantics. An indexed work queue is a work queue that additionally keeps track of the final processing results of work items.

func NewIndexedWorkQueue

func NewIndexedWorkQueue(name string, processor Processor, cfg Config, metricsScope promutils.Scope) (IndexedWorkQueue, error)

Instantiates a new Indexed Work queue.

type Processor

type Processor interface {
	Process(ctx context.Context, workItem WorkItem) (WorkStatus, error)
}

Represents the processor logic to operate on work items.

type WorkItem

type WorkItem interface{}

WorkItem is a generic item that can be stored in the work queue.

type WorkItemID

type WorkItemID = string

type WorkItemInfo

type WorkItemInfo interface {
	Item() WorkItem
	ID() WorkItemID
	Status() WorkStatus
	Error() error
}

Represents the result of the work item processing.

type WorkStatus

type WorkStatus uint8

const (
	WorkStatusNotDone WorkStatus = iota
	WorkStatusSucceeded
	WorkStatusFailed
)

func WorkStatusString

func WorkStatusString(s string) (WorkStatus, error)

WorkStatusString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.

func WorkStatusValues

func WorkStatusValues() []WorkStatus

WorkStatusValues returns all values of the enum

func (WorkStatus) IsAWorkStatus

func (i WorkStatus) IsAWorkStatus() bool

IsAWorkStatus returns "true" if the value is listed in the enum definition. "false" otherwise

func (WorkStatus) IsTerminal

func (w WorkStatus) IsTerminal() bool

func (WorkStatus) String

func (i WorkStatus) String() string