qed

README

Qed

Qed is a simple, Postgres backed task queue.

Getting Started

Setup

Qed leverages Postgres to persist state about running and pending tasks. The Qed schema migration can be found in database/migrations/postgres.

Migrations can be applied directly to a running Postgres instance as follows:

psql -U postgres -h localhost -f database/postgres/schema.sql

Running the Example

A simple usage example can be found here.

The example requires a running Postgres database with the Qed schema migration applied.

To start a new Postgres instance:

docker run -e 'POSTGRES_PASSWORD=password' -d -p 5432:5432 postgres:14.3

Once the Postgres instance is running, the Qed schema migration can be applied:

psql -U postgres -h localhost -f database/postgres/schema.sql

Once the database is set up, the example project can be built and run:

make example

Documentation

Index

• type Options
• type TaskQueue
  • func NewTaskQueue(db *sql.DB, options Options) *TaskQueue
  • func (q *TaskQueue) QueueTask(queue string, data []byte) (string, error)
  • func (q *TaskQueue) QueueTaskWithDelay(queue string, data []byte, delay time.Duration) error
  • func (q *TaskQueue) RegisterHandler(queue string, handler func([]byte) error)
  • func (q *TaskQueue) Run() error

Types

type Options

type Options struct {
	Tick    time.Duration
	Timeout time.Duration
}

type TaskQueue

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

func NewTaskQueue

func NewTaskQueue(db *sql.DB, options Options) *TaskQueue

NewTaskQueue returns a new TaskQueue instance configured to use the specified database connection for persistent task storage. Tasks are polled and dispatched at an interval determined by the `tick` parameter. Tasks which have not been acked after `timeout` are assumed to be blocked and will be retried.

func (*TaskQueue) QueueTask

func (q *TaskQueue) QueueTask(queue string, data []byte) (string, error)

QueueTask queues a new task on the named queue.

func (*TaskQueue) QueueTaskWithDelay

func (q *TaskQueue) QueueTaskWithDelay(queue string, data []byte, delay time.Duration) error

QueueTaskWithDelay queues a new task on the named queue to be run after a specified interval.

func (*TaskQueue) RegisterHandler

func (q *TaskQueue) RegisterHandler(queue string, handler func([]byte) error)

RegisterHandler adds a new handler for the named queue. The handler is invoked when a task is dequeued from the named queue. If a previous handler was registered for the named queue, it will be replaced.

func (*TaskQueue) Run

func (q *TaskQueue) Run() error

Run starts the main queue polling loop.