removed preview notice from JetStream APIs

Makefile

@@ -23,4 +23,4 @@ bundle:
 	deno bundle --log-level info --unstable src/mod.ts ./nats.js
 
 fmt:
-	deno fmt src/ doc/ bin/ nats-base-client/ examples/ tests/
+	deno fmt src/ doc/ bin/ nats-base-client/ examples/ tests/ jetstream.md README.md

jetstream.md

@@ -25,8 +25,7 @@ The new generation of Javascript clients:
 - [nats.ws](https://github.com/nats-io/nats.ws)
 - [nats.deno](https://github.com/nats-io/nats.deno)
 
-all support JetStream, however the functionality is a _preview_, and the APIs
-are subject to change. Please report any issues you find.
+all support JetStream. Please report any issues you find.
 
 ## JetStreamManager
 
@@ -254,6 +253,28 @@ setInterval(() => {
 Note the above example is contrived, as the pull interval is fixed based on some
 interval.
 
+#### Consumer Binding
+
+JetStream's `subscribe()`, and `pullSubscribe()` can `bind` to a specific
+durable consumer. The consumer must already exist, note that if your consumer is
+working on a stream that is sourced from another `bind` is the only way you can
+attach to the correct consumer on the correct stream:
+
+```typescript
+const inbox = createInbox();
+await jsm.consumers.add("A", {
+  durable_name: "me",
+  ack_policy: AckPolicy.None,
+  deliver_subject: inbox,
+});
+
+const opts = consumerOpts();
+opts.bind("A", "me");
+
+const sub = await js.subscribe(subj, opts);
+// process messages...
+```
+
 #### JetStream Queue Consumers
 
 Queue Consumers allow scaling the processing of messages stored in a stream. To
@@ -436,7 +457,7 @@ As the ordered consumer processes messages, it enforces that messages are
 presented to the client with the correct sequence. If a gap is detected, the
 consumer is recreated at the expected sequence.
 
-Most consumer options are rejected, as the ordered consumer has manages its
+Most consumer options are rejected, as the ordered consumer manages its
 configuration in a very specific way.
 
 To create an ordered consumer (assuming a stream that captures `my.messages`):

nats-base-client/jsclient.ts

@@ -69,6 +69,7 @@ import { createInbox } from "./protocol.ts";
 import { headers } from "./headers.ts";
 import { consumerOpts, isConsumerOptsBuilder } from "./jsconsumeropts.ts";
 import { Bucket } from "./kv.ts";
+import { NatsConnectionImpl } from "./nats.ts";
 
 export interface JetStreamSubscriptionInfoable {
   info: JetStreamSubscriptionInfo | null;
@@ -86,6 +87,7 @@ class ViewsImpl implements Views {
   js: JetStreamClientImpl;
   constructor(js: JetStreamClientImpl) {
     this.js = js;
+    jetstreamPreview(this.js.nc);
   }
   async kv(name: string, opts: Partial<KvOptions> = {}): Promise<KV> {
     return Bucket.create(this.js.nc, name, opts);
@@ -737,3 +739,22 @@ function autoAckJsMsg(data: JsMsg | null) {
     data.ack();
   }
 }
+
+const jetstreamPreview = (() => {
+  let once = false;
+  return (nci: NatsConnectionImpl) => {
+    if (!once) {
+      once = true;
+      const { lang } = nci?.protocol?.transport;
+      if (lang) {
+        console.log(
+          `\u001B[33m >> jetstream's materialized views functionality in ${lang} is beta functionality \u001B[0m`,
+        );
+      } else {
+        console.log(
+          `\u001B[33m >> jetstream's materialized views functionality is beta functionality \u001B[0m`,
+        );
+      }
+    }
+  };
+})();

nats-base-client/nats.ts

@@ -284,7 +284,6 @@ export class NatsConnectionImpl implements NatsConnection {
   async jetstreamManager(
     opts: JetStreamOptions = {},
   ): Promise<JetStreamManager> {
-    jetstreamPreview(this);
     const adm = new JetStreamManagerImpl(this, opts);
     try {
       await adm.getAccountInfo();
@@ -301,26 +300,6 @@ export class NatsConnectionImpl implements NatsConnection {
   jetstream(
     opts: JetStreamOptions = {},
   ): JetStreamClient {
-    jetstreamPreview(this);
     return new JetStreamClientImpl(this, opts);
   }
 }
-
-const jetstreamPreview = (() => {
-  let once = false;
-  return (nci: NatsConnectionImpl) => {
-    if (!once) {
-      once = true;
-      const { lang } = nci?.protocol?.transport;
-      if (lang) {
-        console.log(
-          `\u001B[33m >> jetstream functionality in ${lang} is preview functionality \u001B[0m`,
-        );
-      } else {
-        console.log(
-          `\u001B[33m >> jetstream functionality is preview functionality \u001B[0m`,
-        );
-      }
-    }
-  };
-})();

nats-base-client/types.ts

@@ -768,7 +768,6 @@ export interface AccountLimits {
 export interface ConsumerConfig extends ConsumerUpdateConfig {
   "ack_policy": AckPolicy;
   "deliver_policy": DeliverPolicy;
-  "deliver_subject"?: string;
   "deliver_group"?: string;
   "durable_name"?: string;
   "filter_subject"?: string;
@@ -788,6 +787,7 @@ export interface ConsumerUpdateConfig {
   "max_ack_pending"?: number;
   "max_waiting"?: number;
   "headers_only"?: boolean;
+  "deliver_subject"?: string;
 }
 
 export interface Consumer {

tests/jetstream_test.ts

@@ -2759,3 +2759,29 @@ Deno.test("jetstream - bind with diff subject fails", async () => {
   );
   await cleanup(ns, nc);
 });
+
+Deno.test("jetstream - bind example", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}, true));
+  const js = nc.jetstream();
+  const jsm = await nc.jetstreamManager();
+  const subj = `A.*`;
+  await jsm.streams.add({
+    name: "A",
+    subjects: [subj],
+  });
+
+  const inbox = createInbox();
+  await jsm.consumers.add("A", {
+    durable_name: "me",
+    ack_policy: AckPolicy.None,
+    deliver_subject: inbox,
+  });
+
+  const opts = consumerOpts();
+  opts.bind("A", "me");
+
+  const sub = await js.subscribe(subj, opts);
+  assertEquals(sub.getProcessed(), 0);
+
+  await cleanup(ns, nc);
+});