added internal functionality to delete key(s) from a KV (#193)

nats-base-client/kv.ts

@@ -149,6 +149,10 @@ const validKeyRe = /^[-/=.\w]+$/;
 const validSearchKey = /^[-/=.>*\w]+$/;
 const validBucketRe = /^[-\w]+$/;
 
+export interface RemoveKV {
+  remove(k: string): Promise<void>;
+}
+
 export interface RoKV {
   get(k: string): Promise<Entry | null>;
   history(opts?: { key?: string }): Promise<QueuedIterator<Entry>>;
@@ -210,7 +214,7 @@ export function validateBucket(name: string) {
   }
 }
 
-export class Bucket implements KV {
+export class Bucket implements KV, RemoveKV {
   jsm: JetStreamManager;
   js: JetStreamClient;
   stream!: string;
@@ -424,6 +428,43 @@ export class Bucket implements KV {
     return this.jsm.consumers.add(this.stream, opts);
   }
 
+  async remove(k: string): Promise<void> {
+    const ci = await this.consumerOn(k, true);
+    if (ci.num_pending === 0) {
+      await this.jsm.consumers.delete(this.stream, ci.name);
+      return;
+    } else {
+      const ji = this.js as JetStreamClientImpl;
+      const nc = ji.nc;
+      const buf: Promise<boolean>[] = [];
+      const sub = nc.subscribe(ci.config.deliver_subject!, {
+        callback: (err, msg) => {
+          if (err === null) {
+            err = checkJsError(msg);
+          }
+          if (err) {
+            sub.unsubscribe();
+            return;
+          }
+          if (isFlowControlMsg(msg) || isHeartbeatMsg(msg)) {
+            msg.respond();
+            return;
+          }
+          const jm = toJsMsg(msg);
+          buf.push(this.jsm.streams.deleteMessage(this.stream, jm.seq));
+          if (jm.info.pending === 0) {
+            sub.unsubscribe();
+          }
+          jm.ack();
+        },
+      });
+      if (buf.length) {
+        await Promise.all(buf);
+      }
+      await sub.closed;
+    }
+  }
+
   async history(opts: { key?: string } = {}): Promise<QueuedIterator<Entry>> {
     const k = opts.key ?? ">";
     const ci = await this.consumerOn(k, true);

tests/kv_test.ts

@@ -560,3 +560,28 @@ Deno.test("kv - complex key", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("kv - remove key", async () => {
+  const { ns, nc } = await setup(
+    jetstreamServerConf({}, true),
+  );
+  if (await notCompatible(ns, nc)) {
+    return;
+  }
+  const sc = StringCodec();
+  const b = await Bucket.create(nc, nuid.next()) as Bucket;
+
+  await b.put("a.b", sc.encode("ab"));
+  let v = await b.get("a.b");
+  assert(v);
+  assertEquals(sc.decode(v.value), "ab");
+
+  await b.remove("a.b");
+  v = await b.get("a.b");
+  assertEquals(v, null);
+
+  const status = await b.status();
+  assertEquals(status.values, 0);
+
+  await cleanup(ns, nc);
+});