Add connection close listeners and update package versions (#258)

Introduced `ConnectionClosedListener` and related methods for improved resource management on connection closes.

Fix #253

Signed-off-by: Alberto Ricart <[email protected]>

Author: aricart

core/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/nats-core",
-  "version": "3.0.0",
+  "version": "3.0.2-1",
   "exports": {
     ".": "./src/mod.ts",
     "./internal": "./src/internal_mod.ts"

core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/nats-core",
-  "version": "3.0.0",
+  "version": "3.0.2-1",
   "files": [
     "lib/",
     "LICENSE",

core/src/core.ts

@@ -65,6 +65,10 @@ export type SlowConsumerStatus = {
   pending: number;
 };
 
+export type CloseStatus = {
+  type: "close";
+};
+
 export type Status =
   | DisconnectStatus
   | ReconnectStatus
@@ -75,7 +79,8 @@ export type Status =
   | ClientPingStatus
   | StaleConnectionStatus
   | SlowConsumerStatus
-  | ForceReconnectStatus;
+  | ForceReconnectStatus
+  | CloseStatus;
 
 export type MsgCallback<T> = (
   err: Error | null,
@@ -1042,3 +1047,9 @@ export type RequestInfo = {
   client_id?: string;
   nonce?: string;
 };
+
+export type CallbackOptionalErrorFn = (err: Error | void) => void;
+
+export type ConnectionClosedListener = {
+  connectionClosedCallback: CallbackOptionalErrorFn;
+};

core/src/internal_mod.ts

@@ -87,8 +87,10 @@ export type {
   Auth,
   Authenticator,
   CallbackFn,
+  CallbackOptionalErrorFn,
   ClientPingStatus,
   ClusterUpdateStatus,
+  ConnectionClosedListener,
   ConnectionOptions,
   DisconnectStatus,
   Dispatcher,
@@ -129,6 +131,7 @@ export type {
   TokenAuth,
   UserPass,
 } from "./core.ts";
+
 export { createInbox, Match, syncIterator } from "./core.ts";
 export { SubscriptionImpl, Subscriptions } from "./protocol.ts";
 

core/src/nats.ts

@@ -26,6 +26,7 @@ import type { RequestManyOptionsInternal } from "./request.ts";
 import { RequestMany, RequestOne } from "./request.ts";
 
 import type {
+  ConnectionClosedListener,
   ConnectionOptions,
   Context,
   Msg,
@@ -48,6 +49,7 @@ export class NatsConnectionImpl implements NatsConnection {
   options: ConnectionOptions;
   protocol!: ProtocolHandler;
   draining: boolean;
+  closeListeners?: CloseListeners;
 
   private constructor(opts: ConnectionOptions) {
     this.draining = false;
@@ -534,4 +536,50 @@ export class NatsConnectionImpl implements NatsConnection {
     }
     return this.protocol.reconnect();
   }
+
+  // internal
+  addCloseListener(listener: ConnectionClosedListener) {
+    if (this.closeListeners === undefined) {
+      this.closeListeners = new CloseListeners(this.closed());
+    }
+    this.closeListeners.add(listener);
+  }
+  // internal
+  removeCloseListener(listener: ConnectionClosedListener) {
+    if (this.closeListeners) {
+      this.closeListeners.remove(listener);
+    }
+  }
+}
+
+class CloseListeners {
+  listeners: ConnectionClosedListener[];
+
+  constructor(closed: Promise<void | Error>) {
+    this.listeners = [];
+    closed.then((err) => {
+      this.notify(err);
+    });
+  }
+
+  add(listener: ConnectionClosedListener) {
+    this.listeners.push(listener);
+  }
+
+  remove(listener: ConnectionClosedListener) {
+    this.listeners = this.listeners.filter((l) => l !== listener);
+  }
+
+  notify(err: void | Error) {
+    this.listeners.forEach((l) => {
+      if (typeof l.connectionClosedCallback === "function") {
+        try {
+          l.connectionClosedCallback(err);
+        } catch (_) {
+          // ignored
+        }
+      }
+    });
+    this.listeners = [];
+  }
 }

core/src/version.ts

@@ -1,2 +1,2 @@
 // This file is generated - do not edit
-export const version = "3.0.0";
+export const version = "3.0.2-1";

core/tests/basics_test.ts

@@ -35,6 +35,8 @@ import {
   syncIterator,
 } from "../src/internal_mod.ts";
 import type {
+  ConnectionClosedListener,
+  Deferred,
   Msg,
   MsgHdrs,
   MsgImpl,
@@ -1680,3 +1682,47 @@ class MM implements Msg {
     return "";
   }
 }
+
+Deno.test("basics - internal close listener", async () => {
+  const ns = await NatsServer.start();
+  const port = ns.port;
+
+  // nothing bad should happen if none registered
+  let nc = await connect({ port }) as NatsConnectionImpl;
+  await nc.close();
+
+  function makeListener(d: Deferred<unknown>): ConnectionClosedListener {
+    return {
+      connectionClosedCallback: () => {
+        d.resolve();
+      },
+    };
+  }
+
+  // can add and remove
+  nc = await connect({ port }) as NatsConnectionImpl;
+  let done = deferred();
+  let listener = makeListener(done);
+
+  (nc as NatsConnectionImpl).addCloseListener(listener);
+  // @ts-ignore: internal
+  assertEquals((nc as NatsConnectionImpl).closeListeners.listeners.length, 1);
+  (nc as NatsConnectionImpl).removeCloseListener(listener);
+  // @ts-ignore: internal
+  assertEquals((nc as NatsConnectionImpl).closeListeners.listeners.length, 0);
+  await nc.close();
+  done.resolve();
+  await done;
+
+  // closed called
+  nc = await connect({ port }) as NatsConnectionImpl;
+  done = deferred();
+  listener = makeListener(done);
+  (nc as NatsConnectionImpl).addCloseListener(listener);
+  await nc.close();
+  await done;
+  // @ts-ignore: internal
+  assertEquals((nc as NatsConnectionImpl).closeListeners.listeners.length, 0);
+
+  await ns.stop();
+});

jetstream/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/jetstream",
-  "version": "3.0.1-2",
+  "version": "3.0.2-1",
   "exports": {
     ".": "./src/mod.ts",
     "./internal": "./src/internal_mod.ts"
@@ -33,6 +33,6 @@
     "test": "deno test -A --parallel --reload --trace-leaks --quiet tests/ --import-map=import_map.json"
   },
   "imports": {
-    "@nats-io/nats-core": "jsr:@nats-io/[email protected].0"
+    "@nats-io/nats-core": "jsr:@nats-io/[email protected].2-1"
   }
 }

jetstream/import_map.json

@@ -2,8 +2,8 @@
   "imports": {
     "@nats-io/nkeys": "jsr:@nats-io/[email protected]",
     "@nats-io/nuid": "jsr:@nats-io/[email protected]",
-    "@nats-io/nats-core": "jsr:@nats-io/[email protected].0",
-    "@nats-io/nats-core/internal": "jsr:@nats-io/[email protected].0/internal",
+    "@nats-io/nats-core": "jsr:@nats-io/[email protected].2-1",
+    "@nats-io/nats-core/internal": "jsr:@nats-io/[email protected].2-1/internal",
     "test_helpers": "../test_helpers/mod.ts",
     "@std/io": "jsr:@std/[email protected]"
   }

jetstream/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/jetstream",
-  "version": "3.0.1-2",
+  "version": "3.0.2-1",
   "files": [
     "lib/",
     "LICENSE",
@@ -33,7 +33,7 @@
   },
   "description": "jetstream library - this library implements all the base functionality for NATS JetStream for javascript clients",
   "dependencies": {
-    "@nats-io/nats-core": "3.0.0"
+    "@nats-io/nats-core": "3.0.2-1"
   },
   "devDependencies": {
     "@types/node": "^22.13.10",

jetstream/src/consumer.ts

@@ -15,6 +15,7 @@
 
 import type {
   CallbackFn,
+  ConnectionClosedListener,
   Delay,
   MsgImpl,
   QueuedIterator,
@@ -129,6 +130,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   inbox!: string;
   cancelables: Delay[];
   inReset: boolean;
+  closeListener: ConnectionClosedListener;
 
   // callback: ConsumerCallbackFn;
   constructor(
@@ -176,11 +178,15 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     this.abortOnMissingResource = copts.abort_on_missing_resource === true;
     this.bind = copts.bind === true;
 
-    this.consumer.api.nc.closed().then(() => {
-      this._push(() => {
-        this.stop();
-      });
-    });
+    this.closeListener = {
+      // we don't propagate the error here
+      connectionClosedCallback: () => {
+        this._push(() => {
+          this.stop();
+        });
+      },
+    };
+    this.consumer.api.nc.addCloseListener(this.closeListener);
 
     this.start();
   }
@@ -683,6 +689,8 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     if (this.done) {
       return;
     }
+    this.consumer.api.nc.removeCloseListener(this.closeListener);
+
     this.sub?.unsubscribe();
     this.clearTimers();
     this.statusIterator?.stop();

jetstream/tests/consume_test.ts

@@ -611,3 +611,74 @@ Deno.test("consume - consumer not found and no responders", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("consumer - internal close listener", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const nci = nc as NatsConnectionImpl;
+  //@ts-ignore: internal
+  assertEquals(nci.closeListeners, undefined);
+
+  const jsm = await jetstreamManager(nc);
+  await jsm.streams.add({ name: "leak", subjects: ["leak.*"] });
+  const ci = await jsm.consumers.add("leak", {});
+  const js = jsm.jetstream();
+  await js.publish("leak.a");
+  await js.publish("leak.a");
+  await js.publish("leak.a");
+
+  // next
+  const c = await js.consumers.get("leak", ci.name);
+  await c.next();
+  //@ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 0);
+
+  // fetch
+  let iter = await c.fetch({ max_messages: 1 });
+  await (async () => {
+    for await (const _ of iter) {
+      // nothing
+    }
+  })();
+  //@ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 0);
+
+  iter = await c.consume({ max_messages: 1 });
+  await (async () => {
+    for await (const _ of iter) {
+      break;
+    }
+  })();
+  //@ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 0);
+
+  // long
+  iter = await c.consume({ max_messages: 1 });
+  let done = (async () => {
+    for await (const _ of iter) {
+      // nothing
+    }
+  })();
+
+  await nc.flush();
+  //@ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 1);
+  iter.stop();
+  await done;
+  //@ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 0);
+
+  iter = await c.consume({ max_messages: 1 });
+  done = (async () => {
+    for await (const _ of iter) {
+      // nothing
+    }
+  })();
+  // @ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 1);
+  await nc.close();
+  await done;
+  // @ts-ignore: internal
+  assertEquals(nci.closeListeners.listeners.length, 0);
+
+  await cleanup(ns, nc);
+});

kv/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/kv",
-  "version": "3.0.1-1",
+  "version": "3.0.2-1",
   "exports": {
     ".": "./src/mod.ts",
     "./internal": "./src/internal_mod.ts"
@@ -33,7 +33,7 @@
     "test": "deno test -A --parallel --reload --quiet tests/ --import-map=import_map.json"
   },
   "imports": {
-    "@nats-io/nats-core": "jsr:@nats-io/[email protected].0",
-    "@nats-io/jetstream": "jsr:@nats-io/[email protected].1-2"
+    "@nats-io/nats-core": "jsr:@nats-io/[email protected].2-1",
+    "@nats-io/jetstream": "jsr:@nats-io/[email protected].2-1"
   }
 }

kv/import_map.json

@@ -1,9 +1,9 @@
 {
   "imports": {
-    "@nats-io/nats-core": "jsr:@nats-io/[email protected].0",
-    "@nats-io/nats-core/internal": "jsr:@nats-io/nats-core@~3.0.0/internal",
-    "@nats-io/jetstream": "jsr:@nats-io/[email protected].1-2",
-    "@nats-io/jetstream/internal": "jsr:@nats-io/[email protected].1-2/internal",
+    "@nats-io/nats-core": "jsr:@nats-io/[email protected].2-1",
+    "@nats-io/nats-core/internal": "jsr:@nats-io/[email protected].2-1/internal",
+    "@nats-io/jetstream": "jsr:@nats-io/[email protected].2-1",
+    "@nats-io/jetstream/internal": "jsr:@nats-io/[email protected].2-1/internal",
     "test_helpers": "../test_helpers/mod.ts",
     "@nats-io/nkeys": "jsr:@nats-io/[email protected]",
     "@nats-io/nuid": "jsr:@nats-io/[email protected]",