pgmq

A Go client for the PGMQ PostgreSQL message queue extension.

Requires Go 1.27 or later, PGMQ 1.12 or later, and pgx/v5.

Installation

go get github.com/joeychilson/pgmq

Package

• pgmq — typed JSON queues, transactional operations, batch delivery, and concurrent consumers.

Example

Send typed email jobs and consume them with bounded concurrency. Successful jobs are archived; failed jobs become visible again after the visibility timeout.

package main

import (
	"context"
	"errors"
	"log"
	"os"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/joeychilson/pgmq"
)

type Email struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
}

func main() {
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close()

	queue, err := pgmq.New[Email](pool, "emails")
	if err != nil {
		log.Fatal(err)
	}
	if err := queue.Create(ctx); err != nil {
		log.Fatal(err)
	}

	_, err = queue.Send(ctx, Email{
		To:      "user@example.com",
		Subject: "Welcome!",
	})
	if err != nil {
		log.Fatal(err)
	}

	consumer := pgmq.Consumer[Email]{
		Queue:       queue,
		Concurrency: 4,
		Handler: func(ctx context.Context, message pgmq.Message[Email]) error {
			log.Printf("sending %q to %s", message.Payload.Subject, message.Payload.To)
			return nil
		},
	}
	if err := consumer.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
		log.Fatal(err)
	}
}

Development

The integration tests require PGMQ_TEST_DATABASE_URL to point to PostgreSQL with PGMQ available. One local setup is:

docker run -d --name pgmq -e POSTGRES_PASSWORD=postgres -p 5433:5432 ghcr.io/pgmq/pg18-pgmq:v1.12.0
export PGMQ_TEST_DATABASE_URL=postgres://postgres:postgres@localhost:5433/postgres

golangci-lint run ./...
go test ./...
go test -race ./...
go build ./...

Release

Push a semantic version tag in the form vMAJOR.MINOR.PATCH, optionally with a prerelease suffix, to create a GitHub Release with generated notes. The release publishes the Go module source and does not include binary artifacts.

License

MIT