[FIX] migrated 06_heartbeats.js to use the new event name as per jetstream migration docs

aricart · aricart · commit 97a954281469 · 2025-01-07T17:45:55.000-06:00
[CHANGE] changed the examples to be plain javascript, as this makes it easier to use out of the box.

Signed-off-by: Alberto Ricart &lt;alberto@synadia.com&gt;
diff --git a/jetstream/README.md b/jetstream/README.md
@@ -623,11 +623,12 @@ while (true) {
   // watch the to see if the consume operation misses heartbeats
   (async () => {
     for await (const s of await messages.status()) {
-      if (s.type === ConsumerEvents.HeartbeatsMissed) {
+      switch (s.type) {
+        case "heartbeats_missed":
         // you can decide how many heartbeats you are willing to miss
         const n = s.data as number;
-        console.log(`${n} heartbeats missed`);
-        if (n === 2) {
+          console.log(`${s.count} heartbeats missed`);
+          if (s.count === 2) {
           // by calling `stop()` the message processing loop ends
           // in this case this is wrapped by a loop, so it attempts
           // to re-setup the consume
diff --git a/jetstream/examples/01_consumers.js b/jetstream/examples/01_consumers.js
@@ -15,7 +15,7 @@
 
 import { connect } from "@nats-io/transport-deno";
 import { jetstream } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
 const nc = await connect();
diff --git a/jetstream/examples/02_next.js b/jetstream/examples/02_next.js
@@ -15,7 +15,7 @@
 
 import { connect } from "@nats-io/transport-deno";
 import { jetstream } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
 const nc = await connect();
diff --git a/jetstream/examples/03_batch.js b/jetstream/examples/03_batch.js
@@ -15,7 +15,7 @@
 
 import { connect } from "@nats-io/transport-deno";
 import { jetstream } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
 const nc = await connect();
diff --git a/jetstream/examples/04_consume.js b/jetstream/examples/04_consume.js
@@ -15,7 +15,7 @@
 
 import { connect } from "@nats-io/transport-deno";
 import { jetstream } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
 const nc = await connect();
@@ -39,6 +39,6 @@ while (true) {
       m.ack();
     }
   } catch (err) {
-    console.log(`consume failed: ${(err as Error).message}`);
+    console.log(`consume failed: ${err.message}`);
   }
 }
diff --git a/jetstream/examples/05_consume.js b/jetstream/examples/05_consume.js
@@ -15,7 +15,7 @@
 
 import { connect } from "@nats-io/transport-deno";
 import { jetstream } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
 const nc = await connect();
diff --git a/jetstream/examples/06_heartbeats.js b/jetstream/examples/06_heartbeats.js
@@ -14,11 +14,11 @@
  */
 
 import { connect } from "@nats-io/transport-deno";
-import { ConsumerEvents, jetstream } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { jetstream } from "@nats-io/jetstream";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
-const nc = await connect();
+const nc = await connect({ debug: true });
 
 // create a stream with a random name with some messages and a consumer
 const { stream, consumer } = await setupStreamAndConsumer(nc);
@@ -33,16 +33,12 @@ while (true) {
   // watch the to see if the consume operation misses heartbeats
   (async () => {
     for await (const s of messages.status()) {
-      if (s.type === ConsumerEvents.HeartbeatsMissed) {
-        // you can decide how many heartbeats you are willing to miss
-        const n = s.data as number;
-        console.log(`${n} heartbeats missed`);
-        if (n === 2) {
-          // by calling `stop()` the message processing loop ends.
-          // in this case this is wrapped by a loop, so it attempts
-          // to re-setup the consume
-          messages.stop();
-        }
+      switch (s.type) {
+        case "heartbeats_missed":
+          console.log(`${s.count} heartbeats missed`);
+          if (s.count === 2) {
+            messages.stop();
+          }
       }
     }
   })();
diff --git a/jetstream/examples/07_consume_jobs.js b/jetstream/examples/07_consume_jobs.js
@@ -16,8 +16,7 @@
 import { connect, delay } from "@nats-io/transport-deno";
 import { SimpleMutex } from "@nats-io/nats-core/internal";
 import { jetstream } from "@nats-io/jetstream";
-import type { JsMsg } from "@nats-io/jetstream";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 
 // create a connection
 const nc = await connect();
@@ -37,7 +36,7 @@ const messages = await c.consume({ max_messages: 10 });
 // and then only allowing additional processing as others complete
 const rl = new SimpleMutex(5);
 
-async function schedule(m: JsMsg): Promise<void> {
+async function schedule(m) {
   // pretend to do work
   await delay(1000);
   m.ack();
diff --git a/jetstream/examples/08_consume_process.js b/jetstream/examples/08_consume_process.js
@@ -14,7 +14,7 @@
  */
 
 import { connect } from "@nats-io/transport-deno";
-import { setupStreamAndConsumer } from "./util.ts";
+import { setupStreamAndConsumer } from "./util.js";
 import { jetstream } from "@nats-io/jetstream";
 
 // create a connection
@@ -31,7 +31,7 @@ const c = await js.consumers.get(stream, consumer);
 // creating a frequency table based on the subjects found
 const messages = await c.consume();
 
-const data = new Map<string, number>();
+const data = new Map();
 for await (const m of messages) {
   const chunks = m.subject.split(".");
   const v = data.get(chunks[1]) || 0;
diff --git a/jetstream/examples/js_readme_publish_examples.js b/jetstream/examples/js_readme_publish_examples.js
@@ -15,7 +15,6 @@
 
 import { connect, Empty } from "@nats-io/transport-deno";
 import { jetstream, jetstreamManager } from "@nats-io/jetstream";
-import type { PubAck } from "../src/mod.ts";
 
 const nc = await connect();
 const jsm = await jetstreamManager(nc);
@@ -52,7 +51,7 @@ pa = await js.publish("a.b", Empty, { expect: { streamName: "a" } });
 
 // now if you have a stream with different subjects, you can also
 // assert that the last recorded sequence on subject on the stream matches
-const buf: Promise<PubAck>[] = [];
+const buf = [];
 for (let i = 0; i < 100; i++) {
   buf.push(js.publish("a.a", Empty));
 }
diff --git a/jetstream/examples/jsm_readme_jsm_example.js b/jetstream/examples/jsm_readme_jsm_example.js
diff --git a/jetstream/examples/util.js b/jetstream/examples/util.js
@@ -13,18 +13,34 @@
  * limitations under the License.
  */
 
-import { createConsumer, fill, initStream } from "../tests/jstest_util.ts";
-import type { NatsConnection } from "@nats-io/nats-core";
 import { nuid } from "@nats-io/nats-core";
+import { AckPolicy, jetstreamManager } from "@nats-io/jetstream";
 
 export async function setupStreamAndConsumer(
-  nc: NatsConnection,
+  nc,
   messages = 100,
-): Promise<{ stream: string; consumer: string }> {
+) {
   const stream = nuid.next();
-  await initStream(nc, stream, { subjects: [`${stream}.*`] });
-  await fill(nc, stream, messages, { randomize: true });
-  const consumer = await createConsumer(nc, stream);
+  const jsm = await jetstreamManager(nc);
+  const js = jsm.jetstream();
+  await jsm.streams.add({ name: stream, subjects: [`${stream}.*`] });
+  const buf = [];
+  for (let i = 0; i < messages; i++) {
+    buf.push(js.publish(`${stream}.${i}`, `${i}`));
+    if (buf.length % 500 === 0) {
+      await Promise.all(buf);
+      buf.length = 0;
+    }
+  }
+  if (buf.length > 0) {
+    await Promise.all(buf);
+    buf.length = 0;
+  }
+
+  const consumer = await jsm.consumers.add(stream, {
+    name: stream,
+    ack_policy: AckPolicy.Explicit,
+  });
 
   return { stream, consumer };
 }
diff --git a/jetstream/tests/consume_test.ts b/jetstream/tests/consume_test.ts
@@ -14,15 +14,14 @@
  */
 
 import { cleanup, jetstreamServerConf, setup } from "test_helpers";
-import { setupStreamAndConsumer } from "../examples/util.ts";
 import {
   assert,
   assertEquals,
   assertExists,
   assertFalse,
   assertRejects,
 } from "jsr:@std/assert";
-import { initStream } from "./jstest_util.ts";
+import { initStream, setupStreamAndConsumer } from "./jstest_util.ts";
 import {
   deadline,
   deferred,
diff --git a/jetstream/tests/jstest_util.ts b/jetstream/tests/jstest_util.ts
@@ -83,3 +83,15 @@ export function fill(
 
   return Promise.all(a);
 }
+
+export async function setupStreamAndConsumer(
+  nc: NatsConnection,
+  messages = 100,
+): Promise<{ stream: string; consumer: string }> {
+  const stream = nuid.next();
+  await initStream(nc, stream, { subjects: [`${stream}.*`] });
+  await fill(nc, stream, messages, { randomize: true });
+  const consumer = await createConsumer(nc, stream);
+
+  return { stream, consumer };
+}

Original file line number	Diff line number	Diff line change
`@@ -15,7 +15,7 @@`
`15`	`15`
`16`	`16`	`import { connect } from "@nats-io/transport-deno";`
`17`	`17`	`import { jetstream } from "@nats-io/jetstream";`
`18`		`-import { setupStreamAndConsumer } from "./util.ts";`
	`18`	`+import { setupStreamAndConsumer } from "./util.js";`
`19`	`19`
`20`	`20`	`// create a connection`
`21`	`21`	`const nc = await connect();`
`@@ -39,6 +39,6 @@ while (true) {`
`39`	`39`	`m.ack();`
`40`	`40`	`}`
`41`	`41`	`} catch (err) {`
`42`		- console.log(`consume failed: ${(err as Error).message}`);
	`42`	+ console.log(`consume failed: ${err.message}`);
`43`	`43`	`}`
`44`	`44`	`}`