feat(reconnect): support fresh clients rather than reuse

package.json

@@ -66,7 +66,7 @@
     "@typescript-eslint/eslint-plugin": "^8.35.1",
     "@typescript-eslint/parser": "^8.35.1",
     "@typescript-eslint/typescript-estree": "^8.35.1",
-    "chai": "^5.2.0",
+    "chai": "^4.5.0",
     "coveralls-next": "^4.2.1",
     "eslint": "^9.30.1",
     "glob": "^11.0.3",

src/PgPubSub.ts

@@ -296,7 +296,7 @@ export declare interface PgPubSub {
  */
 export class PgPubSub extends EventEmitter {
 
-    public readonly pgClient: PgClient;
+    public pgClient: PgClient;
     public readonly options: PgPubSubOptions;
     public readonly channels: PgChannelEmitter = new PgChannelEmitter();
 
@@ -319,17 +319,34 @@ export class PgPubSub extends EventEmitter {
         this.pgClient = (this.options.pgClient || new Client(this.options)) as
             PgClient;
 
-        this.pgClient.on('end', () => this.emit('end'));
-        this.pgClient.on('error', () => this.emit('error'));
-
         this.onNotification = this.options.executionLock
             ? this.onNotificationLockExec.bind(this)
             : this.onNotification.bind(this)
         ;
         this.reconnect = this.reconnect.bind(this);
         this.onReconnect = this.onReconnect.bind(this);
 
-        this.pgClient.on('notification', this.onNotification);
+        this.initClientListeners(this.pgClient);
+    }
+
+    /**
+     * Sets up event listeners on the given pg client instance.
+     *
+     * @param {PgClient} client - pg client to attach listeners to
+     */
+    private initClientListeners(client: PgClient): void {
+        client.on('end', () => this.emit('end'));
+        client.on('error', (err: Error) => this.emit('error', err));
+        client.on('notification', this.onNotification);
+    }
+
+    /**
+     * Creates a fresh pg.Client instance.
+     *
+     * @return {PgClient}
+     */
+    private createClient(): PgClient {
+        return new Client(this.options) as PgClient;
     }
 
     /**
@@ -645,12 +662,16 @@ export class PgPubSub extends EventEmitter {
 
     /**
      * Reconnect routine, used for implementation of auto-reconnecting db
-     * connection
+     * connection. Creates a fresh pg.Client on each attempt since pg.Client
+     * cannot be reused after disconnect.
      *
      * @access private
      * @return {number}
      */
     private reconnect(): number {
+        this.pgClient.off('end', this.reconnect);
+        this.pgClient.off('error', this.reconnect);
+
         return setTimeout(async () => {
             if (this.options.retryLimit <= ++this.retry) {
                 this.emit('error', new Error(
@@ -660,6 +681,10 @@ export class PgPubSub extends EventEmitter {
                 return this.close();
             }
 
+            this.pgClient.removeAllListeners();
+            this.pgClient = this.createClient();
+            this.initClientListeners(this.pgClient);
+
             this.setOnceHandler(['connect'], this.onReconnect);
 
             try { await this.connect(); } catch (err) { /* ignore */ }

test/mocks/pg.ts

@@ -28,18 +28,44 @@ export interface ClientConfig {
     connectionString?: string;
 }
 
+let _onConnect: ((client: Client) => void) | null = null;
+
+export function setOnConnect(fn: ((client: Client) => void) | null): void {
+    _onConnect = fn;
+}
+
+export function getOnConnect(): ((client: Client) => void) | null {
+    return _onConnect;
+}
+
 // noinspection JSUnusedGlobalSymbols
 export class Client extends EventEmitter {
+    private _connected = false;
+    private _ended = false;
+
     // noinspection JSUnusedGlobalSymbols,JSUnusedLocalSymbols
-    public constructor(options: ClientConfig) {
+    public constructor(_options?: ClientConfig) {
         super();
         this.setMaxListeners(Infinity);
     }
     public connect() {
-        this.emit('connect');
+        if (this._connected || this._ended) {
+            throw new Error(
+                'Client has already been connected. ' +
+                'You cannot reuse a client.',
+            );
+        }
+        this._connected = true;
+
+        if (_onConnect) {
+            _onConnect(this);
+        } else {
+            this.emit('connect');
+        }
     }
     // noinspection JSUnusedGlobalSymbols
     public end() {
+        this._ended = true;
         this.emit('end');
     }
     public async query(queryText: string) {