[FEAT] exported consumer - this allows getting a Consumer without per…

…forming any JSM operations on the server

jetstream/internal_mod.ts

@@ -139,6 +139,7 @@ export type {
   ConsumerMessages,
   ConsumerStatus,
   Expires,
+  ExportedConsumer,
   FetchBytes,
   FetchMessages,
   FetchOptions,

jetstream/jsmstream_api.ts

@@ -41,6 +41,7 @@ import {
 } from "../nats-base-client/core.ts";
 import {
   ApiPagedRequest,
+  ConsumerInfo,
   ExternalStream,
   MsgDeleteRequest,
   MsgRequest,
@@ -60,6 +61,7 @@ import {
 } from "./jsapi_types.ts";
 import {
   Consumer,
+  ExportedConsumer,
   OrderedConsumerOptions,
   OrderedPullConsumerImpl,
   PullConsumerImpl,
@@ -132,6 +134,18 @@ export class ConsumersImpl implements Consumers {
       });
   }
 
+  async bind(
+    stream: string,
+    name: string,
+  ): Promise<ExportedConsumer> {
+    await this.checkVersion();
+    const ci: ConsumerInfo = {
+      stream_name: stream,
+      name: name,
+    } as unknown as ConsumerInfo;
+    return Promise.resolve(new PullConsumerImpl(this.api, ci));
+  }
+
   async ordered(
     stream: string,
     opts?: Partial<OrderedConsumerOptions>,

jetstream/mod.ts

@@ -61,6 +61,7 @@ export type {
   DeliveryInfo,
   Destroyable,
   Expires,
+  ExportedConsumer,
   ExternalStream,
   FetchBytes,
   FetchMessages,

jetstream/tests/consumers_test.ts

@@ -48,6 +48,7 @@ import {
   ConsumerDebugEvents,
   ConsumerEvents,
   ConsumerStatus,
+  PullConsumerImpl,
   PullConsumerMessagesImpl,
 } from "../consumer.ts";
 import { deadline } from "../../nats-base-client/util.ts";
@@ -1105,3 +1106,74 @@ Deno.test("consumers - fetch sync", async () => {
   assertEquals(await sync.next(), null);
   await cleanup(ns, nc);
 });
+
+Deno.test("consumers - bind", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  const js = nc.jetstream();
+  await js.publish("hello");
+  await js.publish("hello");
+
+  await jsm.consumers.add("messages", {
+    durable_name: "a",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+  });
+
+  await jsm.consumers.add("messages", {
+    durable_name: "b",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+  });
+
+  await jsm.consumers.add("messages", {
+    durable_name: "c",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+  });
+
+  jsm.consumers.info = () => {
+    throw new Error("exported is not allowed info");
+  };
+
+  let consumer = await js.consumers.bind("messages", "a");
+  let pci = consumer as PullConsumerImpl;
+  pci.info = () => {
+    throw new Error("exported is not allowed to info");
+  };
+  let iter = await consumer.consume({ max_messages: 2 });
+  for await (const m of iter) {
+    m.ack();
+    if (m.info.pending === 0) {
+      break;
+    }
+  }
+
+  consumer = await js.consumers.bind("messages", "b");
+  pci = consumer as PullConsumerImpl;
+  pci.info = () => {
+    throw new Error("exported is not allowed to info");
+  };
+  iter = await consumer.fetch({ max_messages: 2 });
+  for await (const m of iter) {
+    m.ack();
+  }
+
+  consumer = await js.consumers.bind("messages", "c");
+  pci = consumer as PullConsumerImpl;
+  pci.info = () => {
+    throw new Error("exported is not allowed to info");
+  };
+  assertExists(await consumer.next());
+  assertExists(await consumer.next());
+
+  await cleanup(ns, nc);
+});

jetstream/types.ts

@@ -13,7 +13,11 @@
  * limitations under the License.
  */
 
-import { Consumer, OrderedConsumerOptions } from "./consumer.ts";
+import {
+  Consumer,
+  ExportedConsumer,
+  OrderedConsumerOptions,
+} from "./consumer.ts";
 import {
   JetStreamOptions,
   MsgHdrs,
@@ -462,6 +466,17 @@ export interface Consumers {
     stream: string,
     name?: string | Partial<OrderedConsumerOptions>,
   ): Promise<Consumer>;
+
+  /**
+   * Returns a Consumer configured for the specified stream and consumer name. Note there
+   * are no JSAPI lookups or verifications.
+   * @param stream
+   * @param name
+   */
+  bind(
+    stream: string,
+    name: string,
+  ): Promise<ExportedConsumer>;
 }
 
 export interface ConsumerOpts {