LISTEN/NOTIFY: subscription API, pump with auto-reconnect, pull API, and the connection's callback registration (onNotify, onNotice, ...).
Internal module: not part of the public API. Import the pg_connection hub instead.
Procs
proc listen(conn: PgConnection; channel: string): Future[void] {. ...stackTrace: false, raises: [Exception, ValueError, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Subscribe to channel and start pump. Keeps parked waitNotification across restart; may raise PgTimeoutError (pump stop did not complete in time), PgStateError/PgConnectionError (connection not ready or lost), PgQueryError/PgProtocolError (LISTEN round trip failed), ValueError (channel is empty or contains a NUL byte), or CancelledError (the caller cancelled the call).
proc listenPump(conn: PgConnection): owned(Future[void]) {....stackTrace: false, raises: [Exception], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Background loop: dispatch notifications, auto-reconnect on failure.
proc onListenError(conn: PgConnection; callback: ListenErrorCallback) {. ...raises: [], tags: [], forbids: [].}
- Callback for permanent pump failure. err.transportAlive marks a death the pull API reports as PgListenStoppedError.
proc onNotice(conn: PgConnection; callback: NoticeCallback) {....raises: [], tags: [], forbids: [].}
- Set a callback invoked for each notice/warning the server sends (RAISE NOTICE, RAISE WARNING, deprecation notices).
proc onNotify(conn: PgConnection; callback: NotifyCallback) {....raises: [], tags: [], forbids: [].}
- Set a callback invoked for each incoming NOTIFY message.
proc onNotifyOverflow(conn: PgConnection; callback: NotifyOverflowCallback) {. ...raises: [], tags: [], forbids: [].}
- Set a callback invoked when the pull-API queue overflows; the argument counts what that one arrival discarded (notifyDropped accumulates until the next overflow waitNotification reports).
proc onReconnect(conn: PgConnection; callback: ReconnectCallback) {....raises: [], tags: [], forbids: [].}
-
Set a callback invoked after the listen pump reconnects in place and re-subscribes every channel.
A NOTIFY sent from the drop until the channels are listened to again is lost, as the server queues it only for sessions listening at the time. This callback is the only sign of that gap: re-read here whatever the missed notifications would have announced.
The new session is built from conn.config alone: SET / set_config changes, temporary tables, prepared statements and session advisory locks do not carry over. A SET ROLE is lost too, so later queries on the connection run with the login role's privileges. Put a setting that must survive in extraParams (e.g. ("search_path", "app") or ("role", "app_reader")) instead.
proc reconnectInPlace(conn: PgConnection): owned(Future[void]) {. ...stackTrace: false, raises: [Exception, LibraryError, SslError, ValueError, CatchableError, CancelledError], tags: [RootEffect, WriteIOEffect, TimeEffect], forbids: [].}
- Reconnect and re-LISTEN all channels. Cleans up new transport on failure.
proc startListening(conn: PgConnection) {....raises: [Exception], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Start the notification pump. No-op if one is already running.
proc stopListening(conn: PgConnection): Future[void] {....stackTrace: false, raises: [ Exception, ValueError, LibraryError, SslError, CancelledError, PgTimeoutError], tags: [RootEffect, TimeEffect], forbids: [].}
-
Stop the notification pump, returning the connection to csReady (csClosed if the pump died); may raise PgTimeoutError (pump did not stop in time) or CancelledError (the caller cancelled the stop).
The channels stay subscribed server-side, so notifications queue there and a later listen resumes without losing them — unlike unlisten, which drops the subscription. Use this to run a query on a listening connection, which csListening otherwise rejects.
proc unlisten(conn: PgConnection; channel: string): Future[void] {. ...stackTrace: false, raises: [Exception, ValueError, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Unsubscribe; stops pump if no channels remain (keeps waiter across restart except last). May raise the same errors as listen (PgTimeoutError from the pump stop, PgStateError/PgConnectionError, PgQueryError/PgProtocolError from the UNLISTEN round trip, ValueError for an empty or NUL-bearing name, or CancelledError).
proc waitNotification(conn: PgConnection; timeout: Duration = ZeroDuration): Future[ Notification] {....stackTrace: false, raises: [Exception, ValueError, PgNotifyOverflowError, PgStateError, PgError, PgConnectionError, PgTimeoutError], tags: [RootEffect, TimeEffect], forbids: [].}
- Wait for next notification. Raises PgNotifyOverflowError, PgConnectionError (closed), PgListenError (pump died with the transport), PgListenStoppedError (pump died, connection still up), PgStateError (no pump / concurrent wait / closed by close()), PgTimeoutError (timeout).