[FEAT] allow access to consumer info in watch for internal extenders

nats-base-client/jsclient.ts

@@ -435,7 +435,7 @@ export class JetStreamClientImpl extends BaseApiClient
               );
             }
           }
-
+          jsi.last = info;
           jsi.config = info.config;
           jsi.attached = true;
         }
@@ -504,6 +504,7 @@ export class JetStreamClientImpl extends BaseApiClient
     const ci = await this.api.add(jsi.stream, jsi.config);
     jsi.name = ci.name;
     jsi.config = ci.config;
+    jsi.last = ci;
   }
 
   static ingestionFn(
@@ -630,7 +631,9 @@ class JetStreamSubscriptionImpl extends TypedSubscription<JsMsg>
     const jinfo = this.sub.info as JetStreamSubscriptionInfo;
     const name = jinfo.config.durable_name || jinfo.name;
     const subj = `${jinfo.api.prefix}.CONSUMER.INFO.${jinfo.stream}.${name}`;
-    return await jinfo.api._request(subj) as ConsumerInfo;
+    const ci = await jinfo.api._request(subj) as ConsumerInfo;
+    jinfo.last = ci;
+    return ci;
   }
 }
 
@@ -669,6 +672,7 @@ class JetStreamPullSubscriptionImpl extends JetStreamSubscriptionImpl
 
 interface JetStreamSubscriptionInfo extends ConsumerOpts {
   api: BaseApiClient;
+  last: ConsumerInfo;
   attached: boolean;
   deliver: string;
   bind: boolean;

nats-base-client/kv.ts

@@ -488,6 +488,7 @@ export class Bucket implements KV, KvRemove {
     });
 
     const sub = await this.js.subscribe(subj, copts);
+    qi._data = sub;
     qi.iterClosed.then(() => {
       sub.unsubscribe();
     });

nats-base-client/queued_iterator.ts

@@ -75,6 +75,7 @@ export class QueuedIteratorImpl<T> implements QueuedIterator<T> {
   protocolFilterFn?: ProtocolFilterFn<T>;
   dispatchedFn?: DispatchedFn<T>;
   ctx?: unknown;
+  _data?: unknown; //data is for use by extenders in any way they like
   private err?: Error;
 
   constructor() {

tests/kv_test.ts

@@ -40,6 +40,8 @@ import {
   validateKey,
 } from "../nats-base-client/kv.ts";
 import { notCompatible } from "./helpers/mod.ts";
+import { QueuedIteratorImpl } from "../nats-base-client/queued_iterator.ts";
+import { JetStreamSubscriptionInfoable } from "../nats-base-client/jsclient.ts";
 
 Deno.test("kv - key validation", () => {
   const bad = [
@@ -660,3 +662,28 @@ Deno.test("kv - update key", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("kv - internal consumer", async () => {
+  const { ns, nc } = await setup(
+    jetstreamServerConf({}, true),
+  );
+  if (await notCompatible(ns, nc, "2.6.3")) {
+    return;
+  }
+
+  async function getCount(name: string): Promise<number> {
+    const b = await Bucket.create(nc, name) as Bucket;
+    let watch = await b.watch() as QueuedIteratorImpl<unknown>;
+    const sub = watch._data as JetStreamSubscriptionInfoable;
+    return sub?.info?.last?.num_pending || 0;
+  }
+
+  const name = nuid.next();
+  const b = await Bucket.create(nc, name) as Bucket;
+  assertEquals(await getCount(name), 0);
+
+  await b.put("a", Empty);
+  assertEquals(await getCount(name), 1);
+
+  await cleanup(ns, nc);
+});