chore: add an offset tracking example

Author: magne

example/src/offset_tracking_example.js

@@ -0,0 +1,68 @@
+const rabbit = require("rabbitmq-stream-js-client")
+const { randomUUID } = require("crypto")
+
+const rabbitUser = process.env.RABBITMQ_USER || "rabbit"
+const rabbitPassword = process.env.RABBITMQ_PASSWORD || "rabbit"
+
+async function main() {
+  const streamName = `example-${randomUUID()}`
+  console.log(`Creating stream ${streamName}`)
+
+  const client = await rabbit.connect({
+    hostname: "localhost",
+    port: 5552,
+    username: rabbitUser,
+    password: rabbitPassword,
+    vhost: "/",
+    heartbeat: 0,
+  })
+  await client.createStream({ stream: streamName, arguments: {} })
+  const publisher = await client.declarePublisher({ stream: streamName })
+  const totalMessages = 100
+
+  console.log(`Publishing ${totalMessages} messages`)
+  for (let i = 0; i < totalMessages; i++) {
+    const messageBody = i === totalMessages - 1 ? "marker" : `hello ${i}`
+    await publisher.send(Buffer.from(messageBody))
+  }
+
+  let initialOffset = rabbit.Offset.offset(0n)
+  let firstOffset = initialOffset.value
+  let lastOffset = initialOffset.value
+  let messageCount = 0
+  const consumerRef = "offset-tracking-consumer"
+  const consumer = await client.declareConsumer(
+    { stream: streamName, offset: initialOffset, consumerRef },
+    (message) => {
+      messageCount++
+      if (message.offset === initialOffset.value) {
+        console.log("First message received")
+      }
+      if (messageCount % 10 === 0) {
+        console.log("Storing offset")
+        client.storeOffset({ stream: streamName, reference: consumerRef, offsetValue: message.offset })
+      }
+      if (message.content.toString() === "marker") {
+        console.log("Marker found")
+        client.storeOffset({ stream: streamName, reference: consumerRef, offsetValue: message.offset })
+        lastOffset = message.offset
+      }
+    }
+  )
+
+  console.log(`Start consuming...`)
+  await sleep(2000)
+  console.log(`Done consuming, first offset was ${firstOffset}, last offset was ${lastOffset}`)
+  const lastStoredOffset = await consumer.queryOffset()
+  console.log(`Last stored offset was ${lastStoredOffset}`)
+
+  await client.close()
+}
+
+main()
+  .then(() => console.log("done!"))
+  .catch((res) => {
+    console.log("ERROR ", res)
+    process.exit(-1)
+  })
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms))