Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

queue

package

Go to main page

Versions in this module

v1

v1.1.3

Mar 10, 2026

Changes in this version

+ func WithLogger(ctx context.Context, log *slog.Logger) context.Context

v1.1.2

Jan 13, 2026

v1.1.1

Jan 11, 2026

Changes in this version

+ var ErrInvalidNamespace = errors.New("namespace must not be empty")

+ var ErrReservedNamespace = errors.New("namespace is reserved for system use")

type Manager

+ func (manager *Manager) RegisterQueueWorker(ctx context.Context, meta schema.QueueMeta, worker Worker) (*schema.Queue, error)

+ func (manager *Manager) RegisterTickerWorker(ctx context.Context, meta schema.TickerMeta, worker Worker) (*schema.Ticker, error)

+ func (manager *Manager) RunTickerLoopChan(ctx context.Context, ch chan<- *schema.Ticker, period time.Duration) error

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

type Opt

+ func WithNamespace(name string) Opt

+ func WithTracer(tracer trace.Tracer) Opt

+ type Worker interface

+ Run func(ctx context.Context, payload json.RawMessage) error

v1.1.0

Jan 10, 2026

Changes in this version

+ var ErrInvalidPeriod = errors.New("period must be >= 1ms")

+ var ErrInvalidWorkers = errors.New("workers must be >= 1")

+ type Manager struct

+ func New(ctx context.Context, conn pg.PoolConn, namespace string) (*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) ListNamespaces(ctx context.Context, req schema.NamespaceListRequest) (*schema.NamespaceList, error)

+ func (manager *Manager) ListQueueStatuses(ctx context.Context) ([]schema.QueueStatus, error)

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

+ func (manager *Manager) ListTasks(ctx context.Context, req schema.TaskListRequest) (*schema.TaskList, 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, worker string, queues ...string) (*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) Run(ctx context.Context) error

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

+ func (manager *Manager) RunTickerLoop(ctx context.Context, ch chan<- *schema.Ticker, period time.Duration) error

+ func (manager *Manager) RunTickerLoopNs(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)

+ type Opt func(*opts) error

+ func WithPeriod(d time.Duration) Opt

+ func WithWorkerName(name string) Opt

+ func WithWorkers(n int) Opt

+ type TaskHandler func(context.Context, *schema.Task) error

+ type TickerHandler func(context.Context, *schema.Ticker) error

+ type WorkerPool struct

+ func NewWorkerPool(manager *Manager, opt ...Opt) (*WorkerPool, error)

+ func (wp *WorkerPool) RegisterQueue(ctx context.Context, meta schema.QueueMeta, handler TaskHandler) (*schema.Queue, error)

+ func (wp *WorkerPool) RegisterTicker(ctx context.Context, meta schema.TickerMeta, handler TickerHandler) (*schema.Ticker, error)

+ func (wp *WorkerPool) Run(ctx context.Context) error