feat(jetstream): add watcherPrefix option jetstream/JetStreamManager contexts

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

Author: aricart

jetstream/src/jsbaseclient_api.ts

@@ -15,6 +15,7 @@
 
 import {
   backoff,
+  createInbox,
   delay,
   Empty,
   errors,
@@ -65,6 +66,8 @@ export class BaseApiClientImpl {
 
   constructor(nc: NatsConnection, opts?: JetStreamOptions) {
     this.nc = nc as NatsConnectionImpl;
+    opts = opts || {} as JetStreamOptions;
+    opts.watcherPrefix = opts.watcherPrefix || this.nc.options.inboxPrefix;
     this.opts = defaultJsOptions(opts);
     this._parseOpts();
     this.prefix = this.opts.apiPrefix!;
@@ -85,6 +88,9 @@ export class BaseApiClientImpl {
       prefix = prefix.substr(0, prefix.length - 1);
     }
     this.opts.apiPrefix = prefix;
+
+    // verify that watcherPrefix is valid
+    createInbox(this.opts.watcherPrefix);
   }
 
   async _request(

jetstream/src/jsclient.ts

@@ -18,6 +18,7 @@ import { ConsumerAPIImpl } from "./jsmconsumer_api.ts";
 
 import {
   backoff,
+  createInbox,
   deferred,
   delay,
   Empty,
@@ -174,6 +175,14 @@ export class JetStreamClientImpl extends BaseApiClientImpl
       this.opts,
       { checkAPI },
     ) as JetStreamManagerOptions;
+
+    // fail early if watcherPrefix is bad
+    try {
+      createInbox(opts.watcherPrefix);
+    } catch (err) {
+      return Promise.reject(err);
+    }
+
     return jetstreamManager(this.nc, opts);
   }
 

jetstream/src/jsmstream_api.ts

@@ -177,8 +177,7 @@ export class ConsumersImpl implements Consumers {
     name_prefix = name_prefix || `oc_${nuid.next()}`;
     minValidation("name_prefix", name_prefix);
     deliver_prefix = deliver_prefix ||
-      createInbox((this.api as ConsumerAPIImpl).nc.options.inboxPrefix);
-    minValidation("deliver_prefix", name_prefix);
+      createInbox((this.api as ConsumerAPIImpl).getOptions().watcherPrefix);
 
     const cc = Object.assign({}, opts) as ConsumerConfig;
     cc.ack_policy = AckPolicy.None;

jetstream/src/types.ts

@@ -68,6 +68,12 @@ export type JetStreamOptions = {
    * the default JetStream apiPrefix.
    */
   domain?: string;
+
+  /**
+   * Watcher prefix for inbox subscriptions - these are used for watchers
+   * and push consumers. If not set, it uses ConnectionOptions#inboxPrefix
+   */
+  watcherPrefix?: string;
 };
 
 export type JetStreamManagerOptions = JetStreamOptions & {

jetstream/tests/jetstream_test.ts

@@ -1374,3 +1374,81 @@ Deno.test("jetstream - jsm base timeout", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("jetstream - watcherPrefix", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}));
+
+  async function transitive(js: JetStreamClient): Promise<void> {
+    const { watcherPrefix } = js.getOptions();
+    const jsm = await js.jetstreamManager();
+    assertEquals(watcherPrefix, jsm.getOptions().watcherPrefix);
+
+    const js2 = jsm.jetstream();
+    assertEquals(watcherPrefix, js2.getOptions().watcherPrefix);
+  }
+
+  let js = jetstream(nc);
+  assertEquals(js.getOptions().watcherPrefix, undefined);
+
+  const jsm = await jetstreamManager(nc);
+  assertEquals(jsm.getOptions().watcherPrefix, undefined);
+
+  const nc2 = await connect({ port: ns.port, inboxPrefix: "hello" });
+  js = jetstream(nc2);
+  assertEquals(js.getOptions().watcherPrefix, "hello");
+  await transitive(js);
+
+  js = jetstream(nc2, { watcherPrefix: "bar" });
+  assertEquals(js.getOptions().watcherPrefix, "bar");
+  await transitive(js);
+
+  assertThrows(
+    () => {
+      jetstream(nc2, { watcherPrefix: "hello.*" });
+    },
+    Error,
+    "'prefix' cannot have wildcards ('hello.*')",
+  );
+
+  await assertRejects(
+    () => {
+      return jetstreamManager(nc, { watcherPrefix: "hello.*" });
+    },
+    Error,
+    "'prefix' cannot have wildcards ('hello.*')",
+  );
+
+  await cleanup(ns, nc, nc2);
+});
+
+Deno.test("jetstream - watcher deliver_subject", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}));
+
+  const jsm = await jetstreamManager(nc);
+  await jsm.streams.add({ name: "A", subjects: ["a"] });
+
+  async function assertDeliverTo(
+    js: JetStreamClient,
+    deliverPrefix: string,
+  ): Promise<void> {
+    const pc = await js.consumers.getPushConsumer("A");
+    const { config: { deliver_subject } } = await pc.info(true);
+    assertEquals(deliver_subject?.split(".")[0], deliverPrefix);
+    await pc.delete();
+  }
+
+  // with no prefix, defaults to inbox
+  let js = jetstream(nc);
+  await assertDeliverTo(js, "_INBOX");
+
+  // respects inboxPrefix
+  const nc2 = await connect({ port: ns.port, inboxPrefix: "hallo" });
+  js = jetstream(nc2);
+  await assertDeliverTo(js, "hallo");
+
+  // override inboxPrefix
+  js = jetstream(nc2, { watcherPrefix: "hola" });
+  await assertDeliverTo(js, "hola");
+
+  await cleanup(ns, nc, nc2);
+});

kv/tests/kv_test.ts

@@ -2671,3 +2671,21 @@ Deno.test("kv - encoder", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("kv - watcherPrefix", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}));
+
+  const js = jetstream(nc, { watcherPrefix: "hello" });
+  const kv = await new Kvm(js).create("A");
+
+  const iter = await kv.watch({ key: "a.>" }) as QueuedIteratorImpl<
+    KvWatchEntry
+  >;
+  const pci = iter._data as PushConsumer;
+  const { config: { deliver_subject } } = await pci.info(true);
+  assertEquals(deliver_subject?.split(".")[0], "hello");
+
+  iter.stop();
+
+  await cleanup(ns, nc);
+});

obj/tests/objectstore_test.ts

@@ -34,10 +34,20 @@ import {
   headers,
   nanos,
   nuid,
+  type QueuedIteratorImpl,
 } from "@nats-io/nats-core/internal";
 import type { NatsConnectionImpl } from "@nats-io/nats-core/internal";
-import type { ObjectInfo, ObjectStoreMeta } from "../src/types.ts";
-import { jetstreamManager, StorageType } from "@nats-io/jetstream";
+import type {
+  ObjectInfo,
+  ObjectStoreMeta,
+  ObjectWatchInfo,
+} from "../src/types.ts";
+import {
+  jetstream,
+  jetstreamManager,
+  type PushConsumer,
+  StorageType,
+} from "@nats-io/jetstream/internal";
 import { equals } from "@std/bytes";
 import { digestType, Objm } from "../src/objectstore.ts";
 import { Base64UrlPaddedCodec } from "../src/base64.ts";
@@ -1306,3 +1316,18 @@ Deno.test("os - objm creates right number of replicas", async () => {
   await nc.close();
   await NatsServer.stopAll(servers, true);
 });
+
+Deno.test("objectstore - watcherPrefix", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}));
+  const js = jetstream(nc, { watcherPrefix: "hello" });
+  const objm = new Objm(js);
+  const os = await objm.create("test");
+
+  const watches = await os.watch() as QueuedIteratorImpl<ObjectWatchInfo>;
+  const oc = watches._data as PushConsumer;
+  const { config: { deliver_subject } } = await oc.info(true);
+
+  assertEquals(deliver_subject?.split(".")[0], "hello");
+
+  await cleanup(ns, nc);
+});

runtimes.md

@@ -53,6 +53,38 @@ deno init
 deno add jsr:@nats-io/transport-deno
 ```
 
+## TypeScript Compiler Configuration (Node)
+
+If you are writing your code in TypeScript under node using the `tsc` compiler,
+you will need to configure `tsc` properly so it can find its way around the
+module. You will want a configuration file that looks like this:
+
+```json
+{
+  "compilerOptions": {
+    "target": "esnext",
+    "module": "nodenext",
+    "outDir": "lib/",
+    "moduleResolution": "nodenext",
+    "sourceMap": true,
+    "declaration": true,
+    "allowJs": true,
+    "removeComments": false,
+    "resolveJsonModule": true
+  },
+  "include": [
+    "src/**/*"
+  ],
+  "exclude": [
+    "lib/**/*"
+  ]
+}
+```
+
+Of key importance are the `target`, `module`, `moduleResolution`. The `nodenext`
+identifies that the packages can contain `exports`, and will enable the compiler
+and IDEs to properly resolve imports in your code.
+
 ## Let's add a simple program that we can run
 
 Added to the `nats-dev/<runtime>/index.ts` you used above.