pgmq

README

pgmq-go

A Go (Golang) client for Postgres Message Queue (PGMQ). Based loosely on the Rust client.

Usage

Start a Postgres instance with the PGMQ extension installed:

docker run -d --name postgres -e POSTGRES_PASSWORD=password -p 5432:5432 quay.io/tembo/pgmq-pg:latest

Then

package main

import (
    "context"
    "fmt"

    "github.com/craigpastro/pgmq-go"
)

func main() {
    ctx := context.Background()

    q, err := pgmq.New(ctx, "postgres://postgres:password@localhost:5432/postgres")
    if err != nil {
        panic(err)
    }

    err = q.CreateQueue(ctx, "my_queue")
    if err != nil {
        panic(err)
    }

    id, err := q.Send(ctx, "my_queue", json.RawMessage(`{"foo": "bar"}`))
    if err != nil {
        panic(err)
    }

    msg, err := q.Read(ctx, "my_queue", 30)
    if err != nil {
        panic(err)
    }

    // Archive the message by moving it to the "pgmq.a_<queue_name>" table.
    // Alternatively, you can `Delete` the message, or read and delete in one
    // call by using `Pop`.
    _, err = q.Archive(ctx, "my_queue", id)
    if err != nil {
        panic(err)
    }
}

Contributions

We ❤ contributions.

Documentation

Index

• Variables
• type DB
• type Message
• type PGMQ
  • func New(ctx context.Context, connString string) (*PGMQ, error)
  • func NewFromDB(ctx context.Context, db DB) (*PGMQ, error)
  • func (p *PGMQ) Archive(ctx context.Context, queue string, msgID int64) (bool, error)
  • func (p *PGMQ) ArchiveBatch(ctx context.Context, queue string, msgIDs []int64) ([]int64, error)
  • func (p *PGMQ) Close()
  • func (p *PGMQ) CreateQueue(ctx context.Context, queue string) error
  • func (p *PGMQ) CreateUnloggedQueue(ctx context.Context, queue string) error
  • func (p *PGMQ) Delete(ctx context.Context, queue string, msgID int64) (bool, error)
  • func (p *PGMQ) DeleteBatch(ctx context.Context, queue string, msgIDs []int64) ([]int64, error)
  • func (p *PGMQ) DropQueue(ctx context.Context, queue string) error
  • func (p *PGMQ) Ping(ctx context.Context) error
  • func (p *PGMQ) Pop(ctx context.Context, queue string) (*Message, error)
  • func (p *PGMQ) Read(ctx context.Context, queue string, vt int64) (*Message, error)
  • func (p *PGMQ) ReadBatch(ctx context.Context, queue string, vt int64, numMsgs int64) ([]*Message, error)
  • func (p *PGMQ) Send(ctx context.Context, queue string, msg json.RawMessage) (int64, error)
  • func (p *PGMQ) SendBatch(ctx context.Context, queue string, msgs []json.RawMessage) ([]int64, error)
  • func (p *PGMQ) SendBatchWithDelay(ctx context.Context, queue string, msgs []json.RawMessage, delay int) ([]int64, error)
  • func (p *PGMQ) SendWithDelay(ctx context.Context, queue string, msg json.RawMessage, delay int) (int64, error)

Variables

var ErrNoRows = errors.New("pgmq: no rows in result set")

Types

type DB

type DB interface {
	Ping(ctx context.Context) error
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Close()
}

type Message

type Message struct {
	MsgID      int64
	ReadCount  int64
	EnqueuedAt time.Time
	// VT is "visibility time". The UTC timestamp at which the message will
	// be available for reading again.
	VT      time.Time
	Message json.RawMessage
}

type PGMQ

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

func New

func New(ctx context.Context, connString string) (*PGMQ, error)

New uses the connString to attempt to establish a connection to Postgres. Once a connetion is established it will create the PGMQ extension if it does not already exist.

func NewFromDB

func NewFromDB(ctx context.Context, db DB) (*PGMQ, error)

NewFromDB is a bring your own DB version of New. Given an implementation of DB, it will call Ping to ensure the connection has been established, then create the PGMQ extension if it does not already exist.

func (*PGMQ) Archive

func (p *PGMQ) Archive(ctx context.Context, queue string, msgID int64) (bool, error)

Archive moves a message from the queue table to the archive table by its id. View messages on the archive table with sql:

SELECT * FROM pgmq.a_<queue_name>;

func (*PGMQ) ArchiveBatch

func (p *PGMQ) ArchiveBatch(ctx context.Context, queue string, msgIDs []int64) ([]int64, error)

ArchiveBatch moves a batch of messages from the queue table to the archive table by their ids. View messages on the archive table with sql:

SELECT * FROM pgmq.a_<queue_name>;

func (*PGMQ) Close

func (p *PGMQ) Close()

Close closes the underlying connection pool.

func (*PGMQ) CreateQueue

func (p *PGMQ) CreateQueue(ctx context.Context, queue string) error

CreateQueue creates a new queue. This sets up the queue's tables, indexes, and metadata.

func (*PGMQ) CreateUnloggedQueue

func (p *PGMQ) CreateUnloggedQueue(ctx context.Context, queue string) error

CreateUnloggedQueue creates a new unlogged queue, which uses an unlogged table under the hood. This sets up the queue's tables, indexes, and metadata.

func (*PGMQ) Delete

func (p *PGMQ) Delete(ctx context.Context, queue string, msgID int64) (bool, error)

Delete deletes a message from the queue by its id. This is a permanent delete and cannot be undone. If you want to retain a log of the message, use the Archive method.

func (*PGMQ) DeleteBatch

func (p *PGMQ) DeleteBatch(ctx context.Context, queue string, msgIDs []int64) ([]int64, error)

DeleteBatch deletes a batch of messages from the queue by their ids. This is a permanent delete and cannot be undone. If you want to retain a log of the messages, use the ArchiveBatch method.

func (*PGMQ) DropQueue

func (p *PGMQ) DropQueue(ctx context.Context, queue string) error

DropQueue deletes the given queue. It deletes the queue's tables, indices, and metadata. It will return an error if the queue does not exist.

func (*PGMQ) Ping

func (p *PGMQ) Ping(ctx context.Context) error

Ping calls the underlying Ping function of the DB interface.

func (*PGMQ) Pop

func (p *PGMQ) Pop(ctx context.Context, queue string) (*Message, error)

Pop reads single message from the queue and deletes it at the same time. Similar to Read and ReadBatch if no messages are available an ErrNoRows is returned. Unlike these methods, the visibility timeout does not apply. This is because the message is immediately deleted.

func (*PGMQ) Read

func (p *PGMQ) Read(ctx context.Context, queue string, vt int64) (*Message, error)

Read a single message from the queue. If the queue is empty or all messages are invisible, an ErrNoRows errors is returned. If a message is returned, it is made invisible for the duration of the visibility timeout (vt) in seconds.

func (*PGMQ) ReadBatch

func (p *PGMQ) ReadBatch(ctx context.Context, queue string, vt int64, numMsgs int64) ([]*Message, error)

ReadBatch reads a specified number of messages from the queue. Any messages that are returned are made invisible for the duration of the visibility timeout (vt) in seconds. If vt is 0 it will be set to the default value, vtDefault.

func (*PGMQ) Send

func (p *PGMQ) Send(ctx context.Context, queue string, msg json.RawMessage) (int64, error)

Send sends a single message to a queue. The message id, unique to the queue, is returned.

func (*PGMQ) SendBatch

func (p *PGMQ) SendBatch(ctx context.Context, queue string, msgs []json.RawMessage) ([]int64, error)

SendBatch sends a batch of messages to a queue. The message ids, unique to the queue, are returned.

func (*PGMQ) SendBatchWithDelay

func (p *PGMQ) SendBatchWithDelay(ctx context.Context, queue string, msgs []json.RawMessage, delay int) ([]int64, error)

SendBatchWithDelay sends a batch of messages to a queue with a delay. The delay is specified in seconds. The message ids, unique to the queue, are returned.

func (*PGMQ) SendWithDelay

func (p *PGMQ) SendWithDelay(ctx context.Context, queue string, msg json.RawMessage, delay int) (int64, error)

SendWithDelay sends a single message to a queue with a delay. The delay is specified in seconds. The message id, unique to the queue, is returned.