Documentation
¶
Index ¶
- type MessageQueueImplOpt
- type MessageQueueImplOpts
- type PostgresMessageQueue
- func (p *PostgresMessageQueue) Clone() (func() error, msgqueue.MessageQueue, error)
- func (p *PostgresMessageQueue) IsReady() bool
- func (p *PostgresMessageQueue) SendMessage(ctx context.Context, queue msgqueue.Queue, task *msgqueue.Message) error
- func (p *PostgresMessageQueue) SetQOS(prefetchCount int)
- func (p *PostgresMessageQueue) Subscribe(queue msgqueue.Queue, preAck msgqueue.MsgHandler, postAck msgqueue.MsgHandler) (func() error, error)
- type PubSub
- type PubSubOpt
- type PubSubOpts
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type MessageQueueImplOpt ¶
type MessageQueueImplOpt func(*MessageQueueImplOpts)
func WithLogger ¶
func WithLogger(l *zerolog.Logger) MessageQueueImplOpt
func WithQos ¶
func WithQos(qos int) MessageQueueImplOpt
type MessageQueueImplOpts ¶
type MessageQueueImplOpts struct {
// contains filtered or unexported fields
}
type PostgresMessageQueue ¶
type PostgresMessageQueue struct {
// contains filtered or unexported fields
}
func NewPostgresMQ ¶
func NewPostgresMQ(repo v1.MessageQueueRepository, fs ...MessageQueueImplOpt) (func() error, *PostgresMessageQueue, error)
func (*PostgresMessageQueue) Clone ¶
func (p *PostgresMessageQueue) Clone() (func() error, msgqueue.MessageQueue, error)
func (*PostgresMessageQueue) IsReady ¶
func (p *PostgresMessageQueue) IsReady() bool
func (*PostgresMessageQueue) SendMessage ¶ added in v0.74.3
func (*PostgresMessageQueue) SetQOS ¶
func (p *PostgresMessageQueue) SetQOS(prefetchCount int)
func (*PostgresMessageQueue) Subscribe ¶
func (p *PostgresMessageQueue) Subscribe(queue msgqueue.Queue, preAck msgqueue.MsgHandler, postAck msgqueue.MsgHandler) (func() error, error)
type PubSub ¶ added in v0.98.9
type PubSub struct {
// contains filtered or unexported fields
}
PubSub implements msgqueue.PubSub in Postgres, using LISTEN/NOTIFY for messages less than 8KB and the "MessageQueueItem" table for messages greater than 8KB.
func NewPubSub ¶ added in v0.98.9
NewPubSub creates a new Postgres-backed PubSub over the given message queue repository.
The repo must be built on a dedicated pool from the direct (non-pgbouncer) database URL — never the shared repository pool or the pgbouncer URL. Pub can be called from within durable-write paths (so sharing pools is a deadlock risk), and LISTEN does not survive transaction pooling.
func (*PubSub) IsReady ¶ added in v0.98.9
IsReady reports whether the database backing the pub/sub is reachable. Note that this is informational: it is deliberately not wired into the health probes, since a degraded pub/sub shouldn't fail liveness or readiness.
func (*PubSub) Pub ¶ added in v0.98.9
Pub publishes a message to the topic via NOTIFY. Payloads whose wrapped message exceeds pg_notify's 8KB limit fall back to a short-lived message queue row (see MessageQueueRepository.Notify), which subscribers drain with a ~1s poll — task-stream-event payloads routinely exceed 8KB and stream delivery must not regress.
func (*PubSub) Sub ¶ added in v0.98.9
Sub subscribes to a topic. Inline NOTIFY payloads are handled directly; non-JSON notifications and a 1s ticker wake a poll that drains >8KB fallback rows. Delivery is at-most-once: handler errors are logged, never redelivered.
Fanout divergence: inline NOTIFY payloads reach every subscriber of the topic, but each >8KB fallback row is read and acked by exactly one subscriber, unlike the rabbitmq fanout, which delivers every message to every subscriber.
type PubSubOpt ¶ added in v0.98.9
type PubSubOpt func(*PubSubOpts)
func WithPubSubLogger ¶ added in v0.98.9
func WithPubSubPool ¶ added in v0.98.9
WithPubSubPool provides the pool backing the repository so IsReady can ping the database; without it, IsReady always reports true.
type PubSubOpts ¶ added in v0.98.9
type PubSubOpts struct {
// contains filtered or unexported fields
}
Source Files