pgmq

package module

Go to main page

Versions in this module

v0

v0.6.0

Feb 22, 2025

Changes in this version

+ func Archive(ctx context.Context, db DB, queue string, msgID int64) (bool, error)

+ func ArchiveBatch(ctx context.Context, db DB, queue string, msgIDs []int64) ([]int64, error)

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

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

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

+ func Delete(ctx context.Context, db DB, queue string, msgID int64) (bool, error)

+ func DeleteBatch(ctx context.Context, db DB, queue string, msgIDs []int64) ([]int64, error)

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

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

+ func Send(ctx context.Context, db DB, queue string, msg json.RawMessage) (int64, error)

+ func SendBatch(ctx context.Context, db DB, queue string, msgs []json.RawMessage) ([]int64, error)

+ func SendBatchWithDelay(ctx context.Context, db DB, queue string, msgs []json.RawMessage, delay int) ([]int64, error)

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

+ func SendWithDelay(ctx context.Context, db DB, queue string, msg json.RawMessage, delay int) (int64, error)

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

+ type Database struct

+ Pool *pgxpool.Pool

+ func (d *Database) Init()

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

type Message

+ Headers json.RawMessage

+ func Pop(ctx context.Context, db DB, queue string) (*Message, error)

+ func Read(ctx context.Context, db DB, queue string, vt int64) (*Message, error)

+ func ReadBatch(ctx context.Context, db DB, queue string, vt int64, numMsgs int64) ([]*Message, error)

v0.5.1

Jan 1, 2025

v0.5.0

Nov 7, 2024

Changes in this version

type PGMQ

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

v0.4.0

Nov 10, 2023

Changes in this version

type DB

+ Ping func(ctx context.Context) error

type PGMQ

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

v0.3.0

Nov 8, 2023

Changes in this version

type PGMQ

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

v0.2.0

Sep 27, 2023

Changes in this version

type PGMQ

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

+ func (p *PGMQ) SendWithDelay(ctx context.Context, queue string, msg map[string]any, delay int) (int64, error)

v0.1.0

Sep 4, 2023

Changes in this version

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

+ var ErrPing = errors.New("pgmq: failed to ping db")

+ type DB interface

+ Close func()

+ Exec func(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)

+ Query func(ctx context.Context, sql string, args ...any) (pgx.Rows, error)

+ QueryRow func(ctx context.Context, sql string, args ...any) pgx.Row

+ type Message struct

+ EnqueuedAt time.Time

+ Message map[string]any

+ MsgID int64

+ ReadCount int64

+ VT time.Time

+ type PGMQ struct

+ func MustNew(ctx context.Context, connString string) *PGMQ

+ func New(ctx context.Context, connString string) (*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) ([]bool, error)

+ func (p *PGMQ) Close()

+ func (p *PGMQ) CreateQueue(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) ([]bool, error)

+ func (p *PGMQ) DropQueue(ctx context.Context, queue string) 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 map[string]any) (int64, error)

+ func (p *PGMQ) SendBatch(ctx context.Context, queue string, msgs []map[string]any) ([]int64, error)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL