[FEAT] [JETSTREAM] [KV] support multiple key filters on watchers/history

jetstream/kv.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 The NATS Authors
+ * Copyright 2021-2024 The NATS Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -639,12 +639,16 @@ export class Bucket implements KV, KvRemove {
   }
 
   _buildCC(
-    k: string,
+    k: string | string[],
     content: KvWatchInclude,
     opts: Partial<ConsumerConfig> = {},
   ): Partial<ConsumerConfig> {
-    const ek = this.encodeKey(k);
-    this.validateSearchKey(k);
+    const a = !Array.isArray(k) ? [k] : k;
+    let filter_subjects: string[] | undefined = a.map((k) => {
+      const ek = this.encodeKey(k);
+      this.validateSearchKey(k);
+      return this.fullKeyName(ek);
+    });
 
     let deliver_policy = DeliverPolicy.LastPerSubject;
     if (content === KvWatchInclude.AllHistory) {
@@ -654,10 +658,17 @@ export class Bucket implements KV, KvRemove {
       deliver_policy = DeliverPolicy.New;
     }
 
+    let filter_subject: undefined | string = undefined;
+    if (filter_subjects.length === 1) {
+      filter_subject = filter_subjects[0];
+      filter_subjects = undefined;
+    }
+
     return Object.assign({
       deliver_policy,
       "ack_policy": AckPolicy.None,
-      "filter_subject": this.fullKeyName(ek),
+      filter_subjects,
+      filter_subject,
       "flow_control": true,
       "idle_heartbeat": nanos(5 * 1000),
     }, opts) as Partial<ConsumerConfig>;
@@ -668,7 +679,7 @@ export class Bucket implements KV, KvRemove {
   }
 
   async history(
-    opts: { key?: string; headers_only?: boolean } = {},
+    opts: { key?: string | string[]; headers_only?: boolean } = {},
   ): Promise<QueuedIterator<KvEntry>> {
     const k = opts.key ?? ">";
     const qi = new QueuedIteratorImpl<KvEntry>();

jetstream/tests/kv_test.ts

@@ -301,6 +301,31 @@ Deno.test("kv - history", async () => {
   await cleanup(ns, nc);
 });
 
+Deno.test("kv - history multiple keys", async () => {
+  const { ns, nc } = await setup(
+    jetstreamServerConf({}),
+  );
+  const n = nuid.next();
+  const js = nc.jetstream();
+  const bucket = await js.views.kv(n, { history: 2 });
+
+  await bucket.put("A", Empty);
+  await bucket.put("B", Empty);
+  await bucket.put("C", Empty);
+  await bucket.put("D", Empty);
+
+  const iter = await bucket.history({ key: ["A", "D"] });
+  const buf = [];
+  for await (const e of iter) {
+    buf.push(e.key);
+  }
+
+  assertEquals(buf.length, 2);
+  assertArrayIncludes(buf, ["A", "D"]);
+
+  await cleanup(ns, nc);
+});
+
 Deno.test("kv - cleanups/empty", async () => {
   const { ns, nc } = await setup(
     jetstreamServerConf({}),
@@ -1808,6 +1833,38 @@ Deno.test("kv - watch updates only", async () => {
   await cleanup(ns, nc);
 });
 
+Deno.test("kv - watch multiple keys", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}));
+
+  const js = nc.jetstream();
+  const kv = await js.views.kv("K");
+
+  await kv.put("a", "a");
+  await kv.put("b", "b");
+  await kv.put("c", "c");
+
+  const d = deferred();
+  const iter = await kv.watch({
+    key: ["a", "c"],
+    initializedFn: () => {
+      d.resolve();
+    },
+  });
+
+  const notifications: string[] = [];
+  (async () => {
+    for await (const e of iter) {
+      notifications.push(e.key);
+    }
+  })().then();
+  await d;
+
+  assertEquals(notifications.length, 2);
+  assertArrayIncludes(notifications, ["a", "c"]);
+
+  await cleanup(ns, nc);
+});
+
 Deno.test("kv - watch history", async () => {
   const { ns, nc } = await setup(jetstreamServerConf({}));
 

jetstream/types.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 The NATS Authors
+ * Copyright 2023-2024 The NATS Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -1132,8 +1132,9 @@ export enum KvWatchInclude {
 export type KvWatchOptions = {
   /**
    * A key or wildcarded key following keys as if they were NATS subject names.
+   * Note you can specify multiple keys if running on server 2.10.x or better.
    */
-  key?: string;
+  key?: string | string[];
   /**
    * Notification should only include entry headers
    */
@@ -1173,9 +1174,10 @@ export interface RoKV {
 
   /**
    * Returns an iterator of the specified key's history (or all keys).
+   * Note you can specify multiple keys if running on server 2.10.x or better.
    * @param opts
    */
-  history(opts?: { key?: string }): Promise<QueuedIterator<KvEntry>>;
+  history(opts?: { key?: string | string[] }): Promise<QueuedIterator<KvEntry>>;
 
   /**
    * Returns an iterator that will yield KvEntry updates as they happen.