pgmq

package module
v0.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 22, 2025 License: MIT Imports: 13 Imported by: 1

README

pgmq-go

Go Reference Go Report Card CI codecov

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

pgmq-go works with pgx. The second argument of most functions only needs to satisfy the DB interface, which means it can take, among others, a *pgx.Conn, *pgxpool.Pool, or pgx.Tx.

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()

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

    err = pgmq.CreatePGMQExtension(ctx, pool)
    if err != nil {
        panic(err)
    }

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

    // We can perform various queue operations using a transaction.
    tx, err := pool.Begin(ctx)
    if err != nil {
        panic(err)
    }

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

    msg, err := pgmq.Read(ctx, tx, "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 = pgmq.Archive(ctx, tx, "my_queue", id)
    if err != nil {
        panic(err)
    }

    // Commit the transaction.
    err = tx.Commit(ctx)
    if err != nil {
        panic(err)
    }

    // Close the connection pool.
    pool.Close()
}

Contributions

We ❤ contributions.

See also

Documentation

Index

Constants

This section is empty.

Variables

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

Functions

func Archive added in v0.6.0 

func Archive(ctx context.Context, db DB, 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 ArchiveBatch added in v0.6.0 

func ArchiveBatch(ctx context.Context, db DB, 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 CreatePGMQExtension added in v0.6.0 

func CreatePGMQExtension(ctx context.Context, db DB) error

CreatePGMQExtension will create the PGMQ extension using the provided DB.

func CreateQueue added in v0.6.0 

func CreateQueue(ctx context.Context, db DB, queue string) error

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

func CreateUnloggedQueue added in v0.6.0 

func CreateUnloggedQueue(ctx context.Context, db DB, 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 Delete added in v0.6.0 

func Delete(ctx context.Context, db DB, 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 DeleteBatch added in v0.6.0 

func DeleteBatch(ctx context.Context, db DB, 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 DropQueue added in v0.6.0 

func DropQueue(ctx context.Context, db DB, 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 NewPgxPool added in v0.6.0 

func NewPgxPool(ctx context.Context, connString string) (*pgxpool.Pool, error)

NewPgxPool is a convenience function for creating a new *pgxpool.Pool.

func Send added in v0.6.0 

func Send(ctx context.Context, db DB, 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 SendBatch added in v0.6.0 

func SendBatch(ctx context.Context, db DB, 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 SendBatchWithDelay added in v0.6.0 

func SendBatchWithDelay(ctx context.Context, db DB, 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 SendBatchWithDelayTimestamp added in v0.6.0 

func SendBatchWithDelayTimestamp(ctx context.Context, db DB, queue string, msgs []json.RawMessage, delay time.Time) ([]int64, error)

SendBatchWithDelayTimestamp sends a batch of messages to a queue with a delay. The delay is specified as a timestamp. The message ids, unique to the queue, are returned.

func SendWithDelay added in v0.6.0 

func SendWithDelay(ctx context.Context, db DB, 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.

func SendWithDelayTimestamp added in v0.6.0 

func SendWithDelayTimestamp(ctx context.Context, db DB, queue string, msg json.RawMessage, delay time.Time) (int64, error)

SendWithDelayTimestamp sends a single message to a queue with a delay. The delay is specified as a timestamp. The message id, unique to the queue, is returned. Only supported in pgmq-pg17 and above.

Types

type DB 

type DB interface {
	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
}

type Database added in v0.6.0 

type Database struct {
	Pool *pgxpool.Pool
	// contains filtered or unexported fields
}

func (*Database) Init added in v0.6.0 

func (d *Database) Init()

func (*Database) TestArchive added in v0.6.0 

func (d *Database) TestArchive(t *testing.T)

func (*Database) TestArchiveBatch added in v0.6.0 

func (d *Database) TestArchiveBatch(t *testing.T)

func (*Database) TestArchiveNotExist added in v0.6.0 

func (d *Database) TestArchiveNotExist(t *testing.T)

func (*Database) TestCreateAndDropQueue added in v0.6.0 

func (d *Database) TestCreateAndDropQueue(t *testing.T)

func (*Database) TestCreateUnloggedAndDropQueue added in v0.6.0 

func (d *Database) TestCreateUnloggedAndDropQueue(t *testing.T)

func (*Database) TestDelete added in v0.6.0 

func (d *Database) TestDelete(t *testing.T)

func (*Database) TestDeleteBatch added in v0.6.0 

func (d *Database) TestDeleteBatch(t *testing.T)

func (*Database) TestDeleteNotExist added in v0.6.0 

func (d *Database) TestDeleteNotExist(t *testing.T)

func (*Database) TestDropQueueWhichDoesNotExist added in v0.6.0 

func (d *Database) TestDropQueueWhichDoesNotExist(t *testing.T)

func (*Database) TestPing added in v0.6.0 

func (d *Database) TestPing(t *testing.T)

func (*Database) TestPop added in v0.6.0 

func (d *Database) TestPop(t *testing.T)

func (*Database) TestPopEmptyQueueReturnsNoRows added in v0.6.0 

func (d *Database) TestPopEmptyQueueReturnsNoRows(t *testing.T)

func (*Database) TestRead added in v0.6.0 

func (d *Database) TestRead(t *testing.T)

func (*Database) TestReadBatch added in v0.6.0 

func (d *Database) TestReadBatch(t *testing.T)

func (*Database) TestReadEmptyQueueReturnsNoRows added in v0.6.0 

func (d *Database) TestReadEmptyQueueReturnsNoRows(t *testing.T)

func (*Database) TestSend added in v0.6.0 

func (d *Database) TestSend(t *testing.T)

func (*Database) TestSendAMarshalledStruct added in v0.6.0 

func (d *Database) TestSendAMarshalledStruct(t *testing.T)

func (*Database) TestSendBatch added in v0.6.0 

func (d *Database) TestSendBatch(t *testing.T)

func (*Database) TestSendBatchWithDelayTimestamp added in v0.6.0 

func (d *Database) TestSendBatchWithDelayTimestamp(t *testing.T)

func (*Database) TestSendInvalidJSONFails added in v0.6.0 

func (d *Database) TestSendInvalidJSONFails(t *testing.T)

func (*Database) TestSendWithDelayTimestamp added in v0.6.0 

func (d *Database) TestSendWithDelayTimestamp(t *testing.T)

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
	Headers json.RawMessage // Only supported in pgmq-pg17 and above
}

func Pop added in v0.6.0 

func Pop(ctx context.Context, db DB, 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 Read added in v0.6.0 

func Read(ctx context.Context, db DB, 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 ReadBatch added in v0.6.0 

func ReadBatch(ctx context.Context, db DB, 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.

Source Files

View all Source files

Directories

Path Synopsis
Package mocks is a generated GoMock package.
 Package mocks is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.