PgPubSub class

Implements LISTEN/NOTIFY client for PostgreSQL connections.

It is a basic public interface of this library, so the end-user is going to work with this class directly to solve his/her tasks. Construct it with PgPubSubOptions, subscribe channels once connected, then read messages from either the instance's own 'message' event or the per-channel emitter on PgPubSub.channels.

Signature:

export declare class PgPubSub extends EventEmitter

Extends: EventEmitter

Remarks

Subscribe from inside the 'connect' handler rather than after awaiting connect(). The connection reconnects automatically, and only the handler runs again on each reconnect — subscriptions made once, after the first connect, are not restored.

close() and connect() are a matched pair for temporarily stepping out of message handling: closing releases the channel locks, so another running copy takes over the channels while this one is busy, and connecting again competes for them. Use it when a process needs to do heavy work without holding up its channels. To shut down for good use PgPubSub.destroy() instead, which also removes the listeners and cannot be reconnected.

Example 1

Connect, subscribe two channels and handle their messages:

import { type AnyJson, PgPubSub } from '@imqueue/pg-pubsub';

const pubSub = new PgPubSub({ connectionString: process.env.DB_URL });

// subscribe inside 'connect' so the channels are restored on every reconnect
pubSub.on('connect', async () => {
    await Promise.all(
        ['ChannelOne', 'ChannelTwo'].map(channel => pubSub.listen(channel)),
    );
});

// one handler for every channel...
pubSub.on('message', (channel: string, payload: AnyJson) =>
    console.log(channel, payload),
);

// ...or one per channel
pubSub.channels.on('ChannelOne', (payload: AnyJson) => console.log(1, payload));
pubSub.channels.on('ChannelTwo', (payload: AnyJson) => console.log(2, payload));

await pubSub.connect();

Example 2

Step out of handling and come back, letting another copy take the channels:

await pubSub.close();
// ... heavy work here; another running copy handles the channels meanwhile
await pubSub.connect();

Constructors

Constructor	Modifiers	Description
(constructor)(options, logger)		Constructs a new instance of the `PgPubSub` class

Properties

Property	Modifiers	Type	Description
channels	`readonly`	PgChannelEmitter	Per-channel event emitter. Listening here scopes a handler to one channel, where the instance's own `'message'` event fires for every channel — which is usually what you want when a process listens to several.
logger	`readonly`	AnyLogger	Where connection, listen and lock lifecycle events are reported. Defaults to the console.
options	`readonly`	PgPubSubOptions	Options this instance was constructed with, merged over the defaults in `DefaultOptions`.
pgClient	`readonly`	PgClient	Underlying postgres client. The instance may be replaced during automatic reconnect (pg clients are single-use), so do not cache this reference across reconnects.

Methods

Method	Modifiers	Description
activeChannels()		Returns list of all active subscribed channels
allChannels()		Returns list of all known channels, despite the fact they are listening (active) or not (inactive).
close()		Safely closes this database connection
connect()		Establishes re-connectable database connection
destroy()		Destroys this object properly, destroying all locks, closing all connections and removing all event listeners to avoid memory leaking. So whenever you need to destroy an object programmatically - use this method. Note, that after destroy it is broken and should be removed from memory.
inactiveChannels()		Returns list of all inactive channels (those which are known, but not actively listening at a time)
isActive(channel)		If channel argument passed will return true if channel is in active state (listening by this pub/sub), false - otherwise. If channel is not specified - will return true if there is at least one active channel listened by this pub/sub, false - otherwise.
listen(channel)		Starts listening given channel. If singleListener option is set to true, it guarantees that only one process would be able to listen this channel at a time.
notify(channel, payload)		Performs NOTIFY to a given channel with a given payload to all listening subscribers
on(event, listener)		Sets `'end'` event handler
on(event, listener)		Sets any unknown or user-defined event handler
on(event, listener)		Sets `'connect'` event handler
on(event, listener)		Sets `'close'` event handler
on(event, listener)		Sets `'listen'` event handler
on(event, listener)		Sets `'unlisten'` event handler
on(event, listener)		Sets `'error'` event handler
on(event, listener)		Sets `'reconnect'` event handler
on(event, listener)		Sets `'message'` event handler
on(event, listener)		Sets `'notify'` event handler
once(event, listener)		Sets `'end'` event handler, which fired only one single time
once(event, listener)		Sets any unknown or user-defined event handler, which would fire only one single time
once(event, listener)		Sets `'connect'` event handler, which fired only one single time
once(event, listener)		Sets `'close'` event handler, which fired only one single time
once(event, listener)		Sets `'listen'` event handler, which fired only one single time
once(event, listener)		Sets `'unlisten'` event handler, which fired only one single time
once(event, listener)		Sets `'error'` event handler, which fired only one single time
once(event, listener)		Sets `'reconnect'` event handler, which fired only one single time
once(event, listener)		Sets `'message'` event handler, which fired only one single time
once(event, listener)		Sets `'notify'` event handler, which fired only one single time
unlisten(channel)		Stops listening of the given channel, and, if singleListener option is set to true - will release an acquired lock (if it was settled).
unlistenAll()		Stops listening all connected channels, and, if singleListener option is set to true - will release all acquired locks (if any was settled).

Read this page as plain markdown — no HTML, no navigation. For pasting into an LLM, or for an agent to fetch.