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; what it re-exports is the supported surface (see tests/api_surface.golden).
Procs
proc listen(conn: PgConnection; channel: string): Future[void] {. ...stackTrace: false, raises: [Exception, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Subscribe to channel and start pump. Keeps parked waitNotification across restart; may raise PgTimeoutError.
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.
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.
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, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Unsubscribe; stops pump if no channels remain (keeps waiter across restart except last).
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).