Add getPullConsumerFor(ci) method to get a by Consumer with a ConsumerInfo (bypassing additional lookup)

jetstream/jsmstream_api.ts

@@ -41,6 +41,7 @@ import {
 } from "../nats-base-client/core.ts";
 import {
   ApiPagedRequest,
+  ConsumerInfo,
   ExternalStream,
   MsgDeleteRequest,
   MsgRequest,
@@ -110,6 +111,13 @@ export class ConsumersImpl implements Consumers {
     return Promise.resolve();
   }
 
+  getPullConsumerFor(ci: ConsumerInfo): Consumer {
+    if (ci.config.deliver_subject !== undefined) {
+      throw new Error("push consumer not supported");
+    }
+    return new PullConsumerImpl(this.api, ci);
+  }
+
   async get(
     stream: string,
     name: string | Partial<OrderedConsumerOptions> = {},
@@ -196,6 +204,11 @@ export class StreamImpl implements Stream {
       });
   }
 
+  getConsumerFromInfo(ci: ConsumerInfo): Consumer {
+    return new ConsumersImpl(new ConsumerAPIImpl(this.api.nc, this.api.opts))
+      .getPullConsumerFor(ci);
+  }
+
   getConsumer(
     name?: string | Partial<OrderedConsumerOptions>,
   ): Promise<Consumer> {

jetstream/tests/consumers_test.ts

@@ -19,7 +19,7 @@ import {
   assertRejects,
   assertStringIncludes,
 } from "https://deno.land/[email protected]/assert/mod.ts";
-import { deferred, nanos } from "../../nats-base-client/mod.ts";
+import { deferred, delay, nanos } from "../../nats-base-client/mod.ts";
 import {
   AckPolicy,
   Consumer,
@@ -551,3 +551,97 @@ Deno.test("consumers - inboxPrefix is respected", async () => {
   await done;
   await cleanup(ns, nc);
 });
+
+Deno.test("consumers - getPullConsumerFor", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf(), { inboxPrefix: "x" });
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  const js = nc.jetstream();
+  await js.publish("hello");
+
+  const ci = await jsm.consumers.add("messages", {
+    durable_name: "c",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+  });
+
+  let c = 0;
+  nc.subscribe("$JS.API.CONSUMER.INFO.>", {
+    callback: (_) => {
+      c++;
+    },
+  });
+
+  const consumer = js.consumers.getPullConsumerFor(ci);
+  const iter = await consumer.consume({ bind: true });
+
+  for await (const _m of iter) {
+    break;
+  }
+
+  await nc.flush();
+  assertEquals(c, 0);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("consumers - getPullConsumerFor non existing misses heartbeats", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf(), { inboxPrefix: "x" });
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  const js = nc.jetstream();
+  await js.publish("hello");
+
+  const ci = await jsm.consumers.add("messages", {
+    durable_name: "c",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+    inactive_threshold: nanos(2000),
+  });
+
+  const consumer = js.consumers.getPullConsumerFor(ci);
+  await consumer.delete();
+
+  await delay(1000);
+  await assertRejects(
+    () => {
+      return jsm.consumers.info("messages", "c");
+    },
+    Error,
+    "consumer not found",
+  );
+
+  let c = 0;
+  nc.subscribe("$JS.API.CONSUMER.INFO.>", {
+    callback: (_) => {
+      c++;
+    },
+  });
+
+  const iter = await consumer.consume({ bind: true, idle_heartbeat: 1000 });
+  (async () => {
+    const status = await iter.status();
+    for await (const s of status) {
+      if (s.type === ConsumerEvents.HeartbeatsMissed) {
+        if (s.data === 2) {
+          iter.stop();
+        }
+      }
+    }
+  })().catch();
+
+  for await (const _m of iter) {
+    break;
+  }
+
+  await nc.flush();
+  assertEquals(c, 0);
+
+  await cleanup(ns, nc);
+});

jetstream/types.ts

@@ -467,6 +467,15 @@ export interface Consumers {
     stream: string,
     name?: string | Partial<OrderedConsumerOptions>,
   ): Promise<Consumer>;
+
+  /**
+   * Returns a Consumer based on the ConsumerInfo specified.
+   * Note this method can throw, and it doesn't validate that the
+   * underlying consumer exists. When using a consumer obtained
+   * by this method it is important to check for ConsumerEvents#HeartbeatsMissed
+   * @param info
+   */
+  getPullConsumerFor(info: ConsumerInfo): Consumer;
 }
 
 export interface ConsumerOpts {
@@ -885,6 +894,8 @@ export interface Stream {
     name?: string | Partial<OrderedConsumerOptions>,
   ): Promise<Consumer>;
 
+  getConsumerFromInfo(ci: ConsumerInfo): Consumer;
+
   getMessage(query: MsgRequest): Promise<StoredMsg>;
 
   deleteMessage(seq: number, erase?: boolean): Promise<boolean>;