feat(core): add reconnect logic for specifying servers (#339)

* feat(core): add reconnect logic for specifying servers

- Introduced support for reconnecting to specified servers via an optional `server` parameter in `reconnect()` implementation.

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

Author: aricart

core/src/core.ts

@@ -491,15 +491,22 @@ export interface NatsConnection {
    * - If the reconnection policy given to the client doesn't allow reconnects, the
    * connection will close.
    *
-   * - Messages that are inbound or outbound could  be lost.
+   * - Messages that are inbound or outbound could be lost.
    *
    * - All requests that are in flight will be rejected.
    *
    * Note that the returned promise will reject if the client is already closed, or if
    * it is in the process of draining. If the client is currently disconnected,
    * this API has no effect, as the client is already attempting to reconnect.
+   *
+   * If a server or list of servers is specified, the client will clear its current
+   * server list (the one provided to connect), and set the specified list as the
+   * new server list - previous servers are forgotten, and then attempt reconnects.
+   * If connecting to a TLS secured server, the hostname(s) must match.
+   *
+   * @param server an optional server or list of servers to reconnect to.
    */
-  reconnect(): Promise<void>;
+  reconnect(server?: string | string[]): Promise<void>;
 }
 
 /**

core/src/nats.ts

@@ -554,13 +554,30 @@ export class NatsConnectionImpl implements NatsConnection {
     return this.protocol.features;
   }
 
-  reconnect(): Promise<void> {
+  reconnect(server?: string | string[]): Promise<void> {
     if (this.isClosed()) {
       return Promise.reject(new errors.ClosedConnectionError());
     }
     if (this.isDraining()) {
       return Promise.reject(new errors.DrainingConnectionError());
     }
+    if (server !== undefined) {
+      if (!Array.isArray(server)) {
+        server = [server];
+      }
+      for (const s of server) {
+        if (!s || typeof s !== "string") {
+          return Promise.reject(
+            InvalidArgumentError.format("server", "must be a hostname"),
+          );
+        }
+      }
+
+      this.protocol.servers.clear(true);
+      for (const s of server) {
+        this.protocol.servers.addServer(s);
+      }
+    }
     return this.protocol.reconnect();
   }
 

core/src/servers.ts

@@ -218,8 +218,11 @@ export class Servers {
     this.currentServer = this.servers[0];
   }
 
-  clear(): void {
+  clear(forgetTLS?: boolean): void {
     this.servers.length = 0;
+    if (forgetTLS === true) {
+      this.tlsName = "";
+    }
   }
 
   updateTLSName(): void {

core/tests/reconnect_test.ts

@@ -12,7 +12,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { assert, assertEquals, assertInstanceOf, fail } from "@std/assert";
+import {
+  assert,
+  assertEquals,
+  assertExists,
+  assertInstanceOf,
+  assertRejects,
+  fail,
+} from "@std/assert";
 import { connect } from "./connect.ts";
 import { Lock, NatsServer } from "test_helpers";
 import {
@@ -469,3 +476,114 @@ Deno.test("reconnect - authentication timeout reconnects", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("reconnect - reconnect to new servers", async () => {
+  const ns = await NatsServer.start();
+
+  const nci = await connect({
+    port: ns.port,
+    debug: true,
+  }) as NatsConnectionImpl;
+
+  const done = deferred();
+
+  (async () => {
+    for await (const s of nci.status()) {
+      console.log(s);
+      if (s.type === "reconnect") {
+        if (s.server === "demo.nats.io:4222") {
+          done.resolve();
+        }
+      }
+    }
+  })().then();
+
+  await nci.reconnect("demo.nats.io");
+
+  assert(!nci.isClosed());
+  assertEquals(nci.protocol.servers.getServers().length, 1);
+  const demo = nci.protocol.servers.getServers()[0];
+  assertEquals(demo.hostname, "demo.nats.io");
+  assertEquals(demo.gossiped, false);
+
+  await done;
+  await cleanup(ns, nci);
+});
+
+Deno.test("reconnect - bad args", async () => {
+  const ns = await NatsServer.start();
+  const nc = await connect({ port: ns.port }) as NatsConnectionImpl;
+  const server = nc.protocol.servers.getServers()[0]!;
+
+  async function check(s?: string | string[]): Promise<void> {
+    await assertRejects(
+      () => nc.reconnect(s),
+      Error,
+      "must be a hostname",
+    );
+    assertEquals(nc.protocol.servers.getServers().length, 1);
+    assertExists(nc.protocol.servers.getServers()[0]);
+    assertEquals(nc.protocol.servers.getServers()[0], server);
+  }
+
+  const tests = [
+    ["localhost", "", "demo.nats.io"],
+    [123 as unknown as string],
+    ["localhost", null as unknown as string, "demo.nats.io"],
+    "",
+  ];
+
+  for (const t of tests) {
+    await check(t);
+  }
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("reconnect - accepts valid array", async () => {
+  const ns = await NatsServer.start();
+  const nc = await connect({ port: ns.port }) as NatsConnectionImpl;
+
+  const done = deferred();
+
+  (async () => {
+    for await (const s of nc.status()) {
+      if (s.type === "reconnect") {
+        done.resolve();
+        break;
+      }
+    }
+  })().then();
+
+  await nc.reconnect(["demo.nats.io"]);
+
+  assertEquals(nc.protocol.servers.getServers().length, 1);
+  assertEquals(nc.protocol.servers.getServers()[0].hostname, "demo.nats.io");
+
+  await done;
+  await cleanup(ns, nc);
+});
+
+Deno.test("reconnect - accepts string", async () => {
+  const ns = await NatsServer.start();
+  const nc = await connect({ port: ns.port }) as NatsConnectionImpl;
+
+  const done = deferred();
+
+  (async () => {
+    for await (const s of nc.status()) {
+      if (s.type === "reconnect") {
+        done.resolve();
+        break;
+      }
+    }
+  })().then();
+
+  await nc.reconnect("demo.nats.io");
+
+  assertEquals(nc.protocol.servers.getServers().length, 1);
+  assertEquals(nc.protocol.servers.getServers()[0].hostname, "demo.nats.io");
+
+  await done;
+  await cleanup(ns, nc);
+});