pgqueue

Documentation

Overview

Package pgqueue provides a Postgres-backed queue implementation for scheduling and processing fetch actions. It supports multiple concurrent workers (processes or goroutines)

Index

• type ProcessFunc
• type Queue
  • func New(ctx context.Context, db *database.DB) (*Queue, error)
  • func (q *Queue) Poll(ctx context.Context, workers int, processFunc ProcessFunc)
  • func (q *Queue) ScheduleFetch(ctx context.Context, modulePath, version string, opts *queue.Options) (bool, error)

Types

type ProcessFunc

type ProcessFunc func(ctx context.Context, modulePath, version string) (int, error)

ProcessFunc is the function signature for processing dequeued work.

type Queue

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

Queue implements the Queue interface backed by a Postgres table. It is safe for concurrent use by multiple goroutines and processes.

func New

func New(ctx context.Context, db *database.DB) (*Queue, error)

New creates the queue_tasks table if it doesn't exist and returns a Queue.

func (*Queue) Poll

func (q *Queue) Poll(ctx context.Context, workers int, processFunc ProcessFunc)

Poll starts background polling for work. It spawns the given number of worker goroutines, each of which periodically claims a task, runs processFunc, and deletes the task on completion. It blocks until ctx is cancelled.

func (*Queue) ScheduleFetch

func (q *Queue) ScheduleFetch(ctx context.Context, modulePath, version string, opts *queue.Options) (bool, error)

ScheduleFetch inserts a task into queue_tasks. It returns (true, nil) if the task was inserted, or (false, nil) if it was a duplicate.