queues

type Fair ¶

type Fair struct {
	// contains filtered or unexported fields
}

Fair implements a fair queue where tasks are distributed evenly across owners.

A queue with base key "foo" and owners "owner1" and "owner2" will have the following keys:

• {foo}:queued - set of owners scored by number of queued tasks
• {foo}:active - set of owners scored by number of active tasks
• {foo}:paused - set of paused owners
• {foo}:temp - used internally
• {foo}:o:owner1/0 - e.g. list of tasks for owner1 with priority 0 (low)
• {foo}:o:owner1/1 - e.g. list of tasks for owner1 with priority 1 (high)
• {foo}:o:owner2/0 - e.g. list of tasks for owner2 with priority 0 (low)
• {foo}:o:owner2/1 - e.g. list of tasks for owner2 with priority 1 (high)

Note: it would be nice if owner queues could use distict hash tags and so live on different nodes in a cluster, but our push and pop scripts require atomic changes to the queued/active sets and the task lists.

func NewFair(keyBase string, maxActivePerOwner int) *Fair

func (q *Fair) Done(ctx context.Context, vc valkey.Conn, owner OwnerID) error

func (q *Fair) Dump(ctx context.Context, vc valkey.Conn) ([]byte, error)

func (q *Fair) Pause(ctx context.Context, vc valkey.Conn, owner OwnerID) error

func (q *Fair) Paused(ctx context.Context, vc valkey.Conn) ([]OwnerID, error)

func (q *Fair) Pop(ctx context.Context, vc valkey.Conn) (TaskID, OwnerID, []byte, error)

func (q *Fair) Push(ctx context.Context, vc valkey.Conn, owner OwnerID, priority bool, task []byte) (TaskID, error)

func (q *Fair) Queued(ctx context.Context, vc valkey.Conn) ([]OwnerID, error)

func (q *Fair) Resume(ctx context.Context, vc valkey.Conn, owner OwnerID) error

func (q *Fair) Size(ctx context.Context, vc valkey.Conn, owner OwnerID) (int, error)

type OwnerID ¶ added in v0.19.0

type OwnerID string

OwnerID is the identifier for an owner of tasks in the queue.

type TaskID ¶ added in v0.19.0

type TaskID string

TaskID is the unique identifier for a task in the queue.