pgmq

Dec 19, 2025

Changes in this version

type Consumer

+ func (r *Consumer) Init()

type Producer

+ func (r *Producer) Init()

Mar 2, 2026

Changes in this version

+ type Consumer struct

+ ArchiveAfterHandle bool

+ ConcurrentMsgs int

+ DbPool *sql.DB

+ ExponentialBackoff int

+ ExponentialPollingLimit int

+ MessageHandler MessageHandlerFunc

+ PollingInterval int

+ QueueName string

+ VisibilityTimeout int

+ func NewConsumer(pool *sql.DB, queueName string, handlerFunc MessageHandlerFunc, ...) *Consumer

+ func (s *Consumer) ArchiveMessage(msg *PgmqMessage)

+ func (s *Consumer) DeleteMessage(msg *PgmqMessage)

+ func (s *Consumer) PurgeQueue(msg *PgmqMessage)

+ func (s *Consumer) ShutdownWithWait()

+ func (s *Consumer) Start()

+ type ConsumerModifier func(*Consumer)

+ func WithConcurrentMsgs(count int) ConsumerModifier

+ func WithExponentialBackoff(secs int) ConsumerModifier

+ func WithPollingInterval(secs int) ConsumerModifier

+ func WithVisibilityTimeout(secs int) ConsumerModifier

+ type MessageHandlerFunc func(ctx context.Context, msg *PgmqMessage)

+ type PgmqMessage struct

+ EnqueuedAt *time.Time

+ Headers *string

+ Message *string

+ MsgID int64

+ ReadCount int

+ VT *time.Time

+ type Producer struct

+ DbPool *sql.DB

+ func NewProducer(pool *sql.DB) *Producer

+ func (s *Producer) Produce(queueName, message, headers string)