LISTEN / NOTIFY plumbing.
- onNotify / onListenError / listen / unlisten — channel subscription API.
- startListening / stopListening / listenPump — background pump that converts incoming NotificationResponse messages into queue/ callback dispatch, with auto-reconnect on transport failure.
- reconnectInPlace — replace the dead transport on the existing PgConnection object and re-LISTEN every subscribed channel so external references survive the reconnect.
- waitNotification — async pull entry point with timeout and overflow detection.
Imports lifecycle.connect for reconnectInPlace and simple_query for the LISTEN/UNLISTEN round trips. Re-exported through pg_connection.nim.
Procs
proc listen(conn: PgConnection; channel: string): Future[void] {. ...stackTrace: false, raises: [Exception, CancelledError, PgTimeoutError, ValueError, LibraryError, SslError, PgConnectionError, PgStateError, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Subscribe to a LISTEN channel and start the background notification pump. Propagates PgTimeoutError from stopListening if the previous pump is stuck in a blocking connect(); the connection is then unusable.
proc listenPump(conn: PgConnection): owned(Future[void]) {....stackTrace: false, raises: [Exception], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Background loop: repeatedly receives messages, dispatching notifications. NotificationResponse/NoticeResponse are dispatched inside recvMessage; an asynchronous ErrorResponse (the server terminating this backend) is raised so its diagnostic drives the failure path rather than being silently dropped. Any other non-notification message is discarded. On connection failure, attempts automatic reconnection with exponential backoff (up to listenReconnectMaxAttempts attempts; 0 or negative = unlimited) and re-subscribes to all channels. Exits cleanly when state changes from csListening (via stopListening sending an empty query), then drains until ReadyForQuery. A stop requested while reconnecting (listenStopRequested) is honored at every yield point of the reconnect loop, so stopListening never strands on a pump that would otherwise loop back into csListening after a successful reconnect. The inter-attempt backoff is slept in short ticks that re-check the stop flag, so a stop mid-backoff is observed within a tick instead of after the full interval.
proc onListenError(conn: PgConnection; callback: proc (err: ref PgListenError) {. ...gcsafe, raises: [].}) {....raises: [], tags: [], forbids: [].}
- Set a callback invoked when the listen pump dies permanently (reconnection failed, or the connection was lost with no channels left to re-subscribe). Push API (onNotify) users have no other way to learn the pump is gone; pull API users see the same failure raised from waitNotification.
proc onNotify(conn: PgConnection; callback: NotifyCallback) {....raises: [], tags: [], forbids: [].}
- Set a callback invoked for each incoming NOTIFY message.
proc reconnectInPlace(conn: PgConnection): owned(Future[void]) {. ...stackTrace: false, raises: [Exception, LibraryError, SslError, ValueError, CatchableError, CancelledError], tags: [RootEffect, WriteIOEffect, TimeEffect], forbids: [].}
- Reconnect using stored config, re-LISTENing on all channels. A re-LISTEN failure closes the freshly opened transport so the reconnect never leaks it. A stop observed right after connect() returns discards newConn, so a close() that already tore down the old transport cannot end up with a live socket grafted on after it returned.
proc startListening(conn: PgConnection) {....raises: [Exception], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc stopListening(conn: PgConnection): owned(Future[void]) {....stackTrace: false, raises: [ Exception, CancelledError, PgTimeoutError, ValueError, LibraryError, SslError], tags: [RootEffect, TimeEffect], forbids: [].}
- Stop the background listen pump and return the connection to csReady (or csClosed if the transport died with no live reconnect). listenStopRequested is cleared on every exit that observed the pump stopping. If instead the pump is stuck in a blocking connect() on asyncdispatch and does not stop within listenReconnectStopWaitMs, this raises PgTimeoutError (a PgConnectionError subtype) and leaves the connection csClosed with the flag set; it is then unusable and must be closed and reopened.
proc unlisten(conn: PgConnection; channel: string): Future[void] {. ...stackTrace: false, raises: [Exception, CancelledError, PgTimeoutError, ValueError, LibraryError, SslError, PgConnectionError, PgStateError, CatchableError], tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
- Unsubscribe from a LISTEN channel. Stops the pump if no channels remain. Propagates PgTimeoutError from stopListening (see listen for details).
proc waitNotification(conn: PgConnection; timeout: Duration = ZeroDuration): Future[ Notification] {....stackTrace: false, raises: [Exception, ValueError, PgNotifyOverflowError, PgListenError, PgConnectionError, PgError, PgTimeoutError], tags: [RootEffect, TimeEffect], forbids: [].}
- Wait for the next notification from the buffer. If the buffer is empty, blocks until a notification arrives or timeout expires. Raises PgNotifyOverflowError if notifications were dropped due to queue overflow. Raises PgListenError if the listen pump has died (e.g. reconnection failed).