Versions in this module Expand all Collapse all v0 v0.6.0 Nov 3, 2026 Changes in this version + var ErrNoRows = errors.New("pgmq: no rows in result set") + 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 SendTopic(ctx context.Context, db DB, queue string, msg 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 DB interface + 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 Database struct + Pool *pgxpool.Pool + func (d *Database) Close() + 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) + func (d *Database) TestSetVisibilityTimeout(t *testing.T) + type Message struct + EnqueuedAt time.Time + Headers json.RawMessage + LastReadAt time.Time + Message json.RawMessage + MsgID int64 + ReadCount int64 + VT time.Time + 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) + func SetVisibilityTimeout(ctx context.Context, db DB, queue string, msgID int64, vt int64) (*Message, error)
Go to main page