Class: PostgresTopicBus
Broadcasts structured messages by topic and delivers them to every process listening on the same channel.
CONNECTION SHAPE. Publishing borrows a pooled connection per call, like any
other query. Listening cannot: LISTEN is session state, so the bus holds ONE
dedicated client out of the pool for as long as it has subscribers, no matter
how many topics or listeners are registered. Size the pool with that one
long-lived checkout in mind.
LIFECYCLE. Construction is inert - nothing connects until the first listen (or an explicit start). close is required to give the connection back; a closed bus stays closed and throws on further use rather than silently reconnecting. Register it with the host’s shutdown hook.
FAILURE HANDLING. A lost notification connection reconnects on its own with
bounded backoff while subscribers remain, reporting each failed attempt through
onError. Messages published during the gap are lost - NOTIFY has no replay.
A throwing or rejecting listener never affects the publisher or the other
listeners; its failure goes to onError.
Not safe to share one instance across unrelated channels - construct one bus
per channel. The channel itself is DERIVED from the channel option rather
than taken literally; see PostgresTopicBusOptions.channel and
PostgresTopicBus.channelName.
Constructors
Section titled “Constructors”Constructor
Section titled “Constructor”new PostgresTopicBus(
pool,options?):PostgresTopicBus
Parameters
Section titled “Parameters”options?
Section titled “options?”Returns
Section titled “Returns”PostgresTopicBus
Properties
Section titled “Properties”channelName
Section titled “channelName”
readonlychannelName:string
The resolved Postgres channel this bus listens and publishes on, derived from
the channel option. Read it to confirm two processes agree, or to log what a
set of parts actually resolved to.
Methods
Section titled “Methods”broadcast()
Section titled “broadcast()”broadcast<
TBody>(topic,input):Promise<TopicMessage<TBody>>
Publish a message to topic, returning the envelope that was sent.
Resolves once Postgres has accepted the NOTIFY, which says nothing about
anyone receiving it: sessions not listening at that moment miss it. Publishing
needs no start and no subscribers.
Validation is deliberately front-loaded, since a message that fails on the
wire is far harder to diagnose than one rejected at the call: TypeError for a
blank topic or type, or a body/metadata that would not round-trip through JSON
unchanged (object.isSerializableValue); RangeError when the encoded
envelope exceeds the NOTIFY payload limit, which the automatic metadata
counts against - send a reference and let the receiver fetch the payload.
Throws if the bus is closed.
Type Parameters
Section titled “Type Parameters”TBody extends SerializableValue
Parameters
Section titled “Parameters”string
TopicPublishInput<TBody>
Returns
Section titled “Returns”Promise<TopicMessage<TBody>>
cleanupExpired()
Section titled “cleanupExpired()”cleanupExpired(
options?):Promise<number>
Parameters
Section titled “Parameters”options?
Section titled “options?”limit?
Section titled “limit?”number
scope?
Section titled “scope?”Returns
Section titled “Returns”Promise<number>
close()
Section titled “close()”close():
Promise<void>
Release the notification connection and stop delivering messages. Idempotent.
Cancels any pending reconnect, drops all listeners, then UNLISTENs and
returns the client to the pool. A failed UNLISTEN is reported to the pool as
a release error so the connection is DISCARDED rather than handed to the next
caller still subscribed to the channel. Does not throw; the bus stays closed.
Returns
Section titled “Returns”Promise<void>
history()
Section titled “history()”history<
TBody>(topic,options?):Promise<TopicHistoryPage<TBody>>
Type Parameters
Section titled “Type Parameters”TBody extends SerializableValue
Parameters
Section titled “Parameters”string
options?
Section titled “options?”after?
Section titled “after?”string
limit?
Section titled “limit?”number
scope?
Section titled “scope?”Returns
Section titled “Returns”Promise<TopicHistoryPage<TBody>>
listen()
Section titled “listen()”listen<
TBody>(topic,listener):Promise<() =>Promise<void>>
Subscribe to topic, returning the function that unsubscribes.
Connects on first use. Several listeners may share a topic; each is called once per message. Only messages published AFTER this resolves arrive, so subscribe before triggering whatever you expect to observe.
The returned function removes just this listener and is safe to call twice. The connection stays open once the last listener leaves - close releases it - so a bus that subscribes and unsubscribes per request does not churn connections.
Type Parameters
Section titled “Type Parameters”TBody extends SerializableValue
Parameters
Section titled “Parameters”string
listener
Section titled “listener”TopicListener<TBody>
Returns
Section titled “Returns”Promise<() => Promise<void>>
start()
Section titled “start()”start():
Promise<void>
Open the dedicated notification connection and LISTEN.
Idempotent, and safe to call concurrently - overlapping calls await the same in-flight connect. listen calls this, so it is only needed to surface a connection problem at startup rather than on first subscribe. Throws if the bus is closed, or if the pool cannot hand out a connection.
Returns
Section titled “Returns”Promise<void>