pgqueue

Documentation

Index

• type Manager
  • func NewManager(ctx context.Context, conn pg.PoolConn, opt ...Opt) (*Manager, error)
  • func (manager *Manager) CleanQueue(ctx context.Context, name string) ([]schema.Task, error)
  • func (manager *Manager) Conn() pg.PoolConn
  • func (manager *Manager) CreateTask(ctx context.Context, queue string, meta schema.TaskMeta) (*schema.Task, error)
  • func (manager *Manager) DeleteQueue(ctx context.Context, name string) (*schema.Queue, error)
  • func (manager *Manager) DeleteTicker(ctx context.Context, name string) (*schema.Ticker, error)
  • func (manager *Manager) GetQueue(ctx context.Context, name string) (*schema.Queue, error)
  • func (manager *Manager) GetTicker(ctx context.Context, name string) (*schema.Ticker, error)
  • func (manager *Manager) ListQueues(ctx context.Context, req schema.QueueListRequest) (*schema.QueueList, error)
  • func (manager *Manager) ListTickers(ctx context.Context, req schema.TickerListRequest) (*schema.TickerList, error)
  • func (manager *Manager) Namespace() string
  • func (manager *Manager) NextTask(ctx context.Context, opt ...Opt) (*schema.Task, error)
  • func (manager *Manager) NextTicker(ctx context.Context) (*schema.Ticker, error)
  • func (manager *Manager) NextTickerNs(ctx context.Context, namespace string) (*schema.Ticker, error)
  • func (manager *Manager) RegisterQueue(ctx context.Context, meta schema.QueueMeta) (*schema.Queue, error)
  • func (manager *Manager) RegisterTicker(ctx context.Context, meta schema.TickerMeta) (*schema.Ticker, error)
  • func (manager *Manager) RegisterTickerNs(ctx context.Context, namespace string, meta schema.TickerMeta) (*schema.Ticker, error)
  • func (manager *Manager) ReleaseTask(ctx context.Context, task uint64, success bool, result any, status *string) (*schema.Task, error)
  • func (manager *Manager) RunNotificationLoop(parent context.Context, ch chan<- *pg.Notification) error
  • func (manager *Manager) RunTaskLoop(ctx context.Context, ch chan<- *schema.Task) error
  • func (manager *Manager) RunTickerLoop(ctx context.Context, namespace string, ch chan<- *schema.Ticker) error
  • func (manager *Manager) UpdateQueue(ctx context.Context, name string, meta schema.QueueMeta) (*schema.Queue, error)
  • func (manager *Manager) UpdateTicker(ctx context.Context, name string, meta schema.TickerMeta) (*schema.Ticker, error)
  • func (manager *Manager) Worker() string
• type Opt
  • func OptNamespace(v string) Opt
  • func OptWorker(v string) Opt
• type TaskPool
  • func NewTaskPool(workers uint) *TaskPool
  • func (pool *TaskPool) Close()
  • func (pool *TaskPool) RunTask(ctx context.Context, t *schema.Task, fn server.PGCallback, result func(error))
  • func (pool *TaskPool) RunTicker(ctx context.Context, t *schema.Ticker, fn server.PGCallback, ...)

Types

type Manager

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

func NewManager

func NewManager(ctx context.Context, conn pg.PoolConn, opt ...Opt) (*Manager, error)

func (*Manager) CleanQueue

func (manager *Manager) CleanQueue(ctx context.Context, name string) ([]schema.Task, error)

CleanQueue removes stale tasks from a queue, and returns the tasks removed

func (*Manager) Conn

func (manager *Manager) Conn() pg.PoolConn

func (*Manager) CreateTask

func (manager *Manager) CreateTask(ctx context.Context, queue string, meta schema.TaskMeta) (*schema.Task, error)

CreateTask creates a new task, and returns it.

func (*Manager) DeleteQueue

func (manager *Manager) DeleteQueue(ctx context.Context, name string) (*schema.Queue, error)

DeleteQueue deletes an existing queue, and returns it

func (*Manager) DeleteTicker

func (manager *Manager) DeleteTicker(ctx context.Context, name string) (*schema.Ticker, error)

DeleteTicker deletes an existing ticker, and returns the deleted ticker.

func (*Manager) GetQueue

func (manager *Manager) GetQueue(ctx context.Context, name string) (*schema.Queue, error)

GetQueue returns a queue by name

func (*Manager) GetTicker

func (manager *Manager) GetTicker(ctx context.Context, name string) (*schema.Ticker, error)

GetTicker returns a ticker by name

func (*Manager) ListQueues

func (manager *Manager) ListQueues(ctx context.Context, req schema.QueueListRequest) (*schema.QueueList, error)

ListQueues returns all queues in a namespace as a list

func (*Manager) ListTickers

func (manager *Manager) ListTickers(ctx context.Context, req schema.TickerListRequest) (*schema.TickerList, error)

ListTickers returns all tickers in a namespace as a list

func (*Manager) Namespace

func (manager *Manager) Namespace() string

func (*Manager) NextTask

func (manager *Manager) NextTask(ctx context.Context, opt ...Opt) (*schema.Task, error)

NextTask retains a task, and returns it. Returns nil if there is no task to retain

func (*Manager) NextTicker

func (manager *Manager) NextTicker(ctx context.Context) (*schema.Ticker, error)

NextTicker returns the next matured ticker, or nil

func (*Manager) NextTickerNs

func (manager *Manager) NextTickerNs(ctx context.Context, namespace string) (*schema.Ticker, error)

NextTickerNs returns the next matured ticker in a namespace, or nil

func (*Manager) RegisterQueue

func (manager *Manager) RegisterQueue(ctx context.Context, meta schema.QueueMeta) (*schema.Queue, error)

RegisterQueue creates a new queue, or updates an existing queue, and returns it.

func (*Manager) RegisterTicker

func (manager *Manager) RegisterTicker(ctx context.Context, meta schema.TickerMeta) (*schema.Ticker, error)

RegisterTicker creates a new ticker, or updates an existing ticker, and returns it.

func (*Manager) RegisterTickerNs

func (manager *Manager) RegisterTickerNs(ctx context.Context, namespace string, meta schema.TickerMeta) (*schema.Ticker, error)

RegisterTicker creates a new ticker, or updates an existing ticker, and returns it.

func (*Manager) ReleaseTask

func (manager *Manager) ReleaseTask(ctx context.Context, task uint64, success bool, result any, status *string) (*schema.Task, error)

ReleaseTask releases a task from a queue, and returns it. Can optionally set the status

func (*Manager) RunNotificationLoop

func (manager *Manager) RunNotificationLoop(parent context.Context, ch chan<- *pg.Notification) error

RunNotificationLoop runs a loop to process database notifications, until the context is cancelled or an error occurs.

func (*Manager) RunTaskLoop

func (manager *Manager) RunTaskLoop(ctx context.Context, ch chan<- *schema.Task) error

RunTaskLoop runs a loop to process tasks, until the context is cancelled or an error occurs.

func (*Manager) RunTickerLoop

func (manager *Manager) RunTickerLoop(ctx context.Context, namespace string, ch chan<- *schema.Ticker) error

RunTickerLoop runs a loop to process matured tickers in a namespace, until the context is cancelled, or an error occurs.

func (*Manager) UpdateQueue

func (manager *Manager) UpdateQueue(ctx context.Context, name string, meta schema.QueueMeta) (*schema.Queue, error)

UpdateQueue updates an existing queue, and returns it.

func (*Manager) UpdateTicker

func (manager *Manager) UpdateTicker(ctx context.Context, name string, meta schema.TickerMeta) (*schema.Ticker, error)

UpdateTicker updates an existing ticker, and returns it.

func (*Manager) Worker

func (manager *Manager) Worker() string

type Opt

type Opt func(*opt) error

Opt represents a function that modifies the options

func OptNamespace

func OptNamespace(v string) Opt

Set the namespace for the tickers and queues

func OptWorker

func OptWorker(v string) Opt

Set the worker name when a task is locked for work

type TaskPool

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

func NewTaskPool

func NewTaskPool(workers uint) *TaskPool

func (*TaskPool) Close

func (pool *TaskPool) Close()

func (*TaskPool) RunTask

func (pool *TaskPool) RunTask(ctx context.Context, t *schema.Task, fn server.PGCallback, result func(error))

Queue a task for running. Will block whilst all the workers are busy

func (*TaskPool) RunTicker

func (pool *TaskPool) RunTicker(ctx context.Context, t *schema.Ticker, fn server.PGCallback, result func(error))

Queue a ticker for running. Will block whilst all the workers are busy