feat: expose query offset in the client (#207)

* feat: expose query offset in the client

* fix: refactor of test

* fix: remove only

---------

Co-authored-by: magne <[email protected]>

Author: l4mby

src/client.ts

@@ -265,6 +265,10 @@ export class Client {
     })
   }
 
+  public queryOffset(params: QueryOffsetParams) {
+    return this.connection.queryOffset(params)
+  }
+
   private async closeAllConsumers(manuallyClose: boolean) {
     await Promise.all([...this.consumers.values()].map(({ consumer }) => consumer.close(manuallyClose)))
     this.consumers = new Map<string, ConsumerMappedValue>()

test/e2e/offset.test.ts

@@ -204,23 +204,40 @@ describe("offset", () => {
   })
 
   describe("query", () => {
-    it("the consumer is able to track the offset of the stream through queryOffset method", async () => {
-      let offset: bigint = 0n
+    it("the consumer is able to track the offset and start from the stored offset", async () => {
+      const consumerOneMessages: Message[] = []
+      const consumerTwoMessages: Message[] = []
+      const publisher = await client.declarePublisher({ stream: testStreamName })
+      await publisher.send(Buffer.from("hello"))
+      await publisher.send(Buffer.from("marker"))
+      await publisher.send(Buffer.from("hello"))
+      await publisher.send(Buffer.from("world"))
+
       const consumer = await client.declareConsumer(
-        { stream: testStreamName, offset: Offset.next(), consumerRef: "my_consumer" },
+        { stream: testStreamName, offset: Offset.first(), consumerRef: "my_consumer" },
         async (message: Message) => {
-          await consumer.storeOffset(message.offset!)
-          offset = message.offset!
+          consumerOneMessages.push(message)
+          if (message.content.toString() === "marker") {
+            await consumer.storeOffset(message.offset!)
+          }
+        }
+      )
+      await eventually(async () => {
+        const [foundMarker] = consumerOneMessages.filter((msg) => msg.content.toString() === "marker")
+        expect(foundMarker).to.not.be.undefined
+        const offset = await client.queryOffset({ stream: testStreamName, reference: "my_consumer" })
+        expect(offset).to.not.be.undefined
+      }, 3000)
+      const storedOffset = await client.queryOffset({ stream: testStreamName, reference: "my_consumer" })
+      await client.declareConsumer(
+        { stream: testStreamName, offset: Offset.offset(storedOffset), consumerRef: "my_consumer" },
+        async (message: Message) => {
+          consumerTwoMessages.push(message)
         }
       )
-      const publisher = await client.declarePublisher({ stream: testStreamName })
-
-      await publisher.send(Buffer.from("hello"))
-      await publisher.send(Buffer.from("world"))
 
       await eventually(async () => {
-        const result = await consumer.queryOffset()
-        expect(result).eql(offset)
+        expect(consumerTwoMessages.length).eql(3)
       })
     }).timeout(10000)