chore: fix offset tracking example

Author: magne

example/src/offset_tracking_example.js

@@ -17,43 +17,48 @@ async function main() {
   })
   await client.createStream({ stream: streamName, arguments: {} })
   const publisher = await client.declarePublisher({ stream: streamName })
-  const messageCount = 100
+  const toSend = 100
 
-  console.log(`Publishing ${messageCount} messages`)
-  for (let i = 0; i < messageCount; i++) {
-    const body = i === messageCount - 1 ? "marker" : `hello ${i}`
+  console.log(`Publishing ${toSend} messages`)
+  for (let i = 0; i < toSend; i++) {
+    const body = i === toSend - 1 ? "marker" : `hello ${i}`
     await publisher.send(Buffer.from(body))
   }
 
-  const startFrom = rabbit.Offset.offset(0n)
-  let firstOffset = startFrom.value
-  let lastOffset = startFrom.value
-  let messageReceivedCount = 0
   const consumerRef = "offset-tracking-tutorial"
-  const consumer = await client.declareConsumer({ stream: streamName, offset: startFrom, consumerRef }, (message) => {
-    messageReceivedCount++
-    if (message.offset === startFrom.value) {
-      console.log("First message received")
-      firstOffset = message.offset
-    }
-    if (messageReceivedCount % 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
+  let firstOffset = undefined
+  let offsetSpecification = rabbit.Offset.first()
+  try {
+    const offset = await client.queryOffset({ reference: consumerRef, stream: streamName })
+    offsetSpecification = rabbit.Offset.offset(offset + 1n)
+  } catch (e) {}
+
+  let lastOffset = offsetSpecification.value
+  let messageCount = 0
+  const consumer = await client.declareConsumer(
+    { stream: streamName, offset: offsetSpecification, consumerRef },
+    async (message) => {
+      messageCount++
+      if (!firstOffset && messageCount === 1) {
+        firstOffset = message.offset
+        console.log("First message received")
+      }
+      if (messageCount % 10 === 0) {
+        await consumer.storeOffset(message.offset)
+      }
+      if (message.content.toString() === "marker") {
+        console.log("Marker found")
+        lastOffset = message.offset
+        await consumer.storeOffset(message.offset)
+        console.log(`Done consuming, first offset was ${firstOffset}, last offset was ${lastOffset}`)
+        await consumer.close(true)
+        process.exit(0)
+      }
     }
-  })
+  )
 
   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()