StreamStats Command

src/connection.ts

@@ -41,6 +41,8 @@ import { Consumer, ConsumerFunc } from "./consumer"
 import { UnsubscribeResponse } from "./responses/unsubscribe_response"
 import { UnsubscribeRequest } from "./requests/unsubscribe_request"
 import { CreditRequest, CreditRequestParams } from "./requests/credit_request"
+import { StreamStatsRequest } from "./requests/stream_stats_request"
+import { StreamStatsResponse } from "./responses/stream_stats_response"
 import { DeliverResponse } from "./responses/deliver_response"
 import { QueryOffsetResponse } from "./responses/query_offset_response"
 import { QueryOffsetRequest } from "./requests/query_offset_request"
@@ -266,6 +268,15 @@ export class Connection {
     return res.sequence
   }
 
+  public async streamStatsRequest(streamName: string) {
+    const res = await this.sendAndWait<StreamStatsResponse>(new StreamStatsRequest(streamName))
+    if (!res.ok) {
+      throw new Error(`Stream Stats command returned error with code ${res.code} - ${errorMessageOf(res.code)}`)
+    }
+    this.logger.info(`Statistics for stream name ${streamName}, ${res.statistics}`)
+    return res.statistics
+  }
+
   public async queryOffset(params: QueryOffsetParams): Promise<bigint> {
     this.logger.debug(`Query Offset...`)
     const res = await this.sendAndWait<QueryOffsetResponse>(new QueryOffsetRequest(params))

src/requests/stream_stats_request.ts

@@ -0,0 +1,16 @@
+import { StreamStatsResponse } from "../responses/stream_stats_response"
+import { AbstractRequest } from "./abstract_request"
+import { DataWriter } from "./data_writer"
+
+export class StreamStatsRequest extends AbstractRequest {
+  readonly responseKey = StreamStatsResponse.key
+  readonly key = 0x001c
+
+  constructor(private streamName: string) {
+    super()
+  }
+
+  writeContent(writer: DataWriter) {
+    writer.writeString(this.streamName)
+  }
+}

src/response_decoder.ts

@@ -37,6 +37,7 @@ import { Message, MessageApplicationProperties, MessageProperties } from "./prod
 import { ApplicationProperties } from "./amqp10/applicationProperties"
 import { TuneResponse } from "./responses/tune_response"
 import { PublishErrorResponse } from "./responses/publish_error_response"
+import { StreamStatsResponse } from "./responses/stream_stats_response"
 import { StoreOffsetResponse } from "./responses/store_offset_response"
 import { QueryOffsetResponse } from "./responses/query_offset_response"
 
@@ -474,6 +475,7 @@ export class ResponseDecoder {
     this.addFactoryFor(QueryPublisherResponse)
     this.addFactoryFor(SubscribeResponse)
     this.addFactoryFor(UnsubscribeResponse)
+    this.addFactoryFor(StreamStatsResponse)
     this.addFactoryFor(StoreOffsetResponse)
     this.addFactoryFor(QueryOffsetResponse)
   }

src/responses/stream_stats_response.ts

@@ -0,0 +1,34 @@
+import { AbstractResponse } from "./abstract_response"
+import { RawResponse } from "./raw_response"
+
+export interface Statistics {
+  committedChunkId: bigint
+  firstChunkId: bigint
+  lastChunkId: bigint
+}
+
+export class StreamStatsResponse extends AbstractResponse {
+  static key = 0x801c
+  private rawStats: Record<string, bigint> = {}
+  readonly statistics: Statistics = {
+    committedChunkId: BigInt(0),
+    firstChunkId: BigInt(0),
+    lastChunkId: BigInt(0),
+  }
+
+  constructor(response: RawResponse) {
+    super(response)
+    this.verifyKey(StreamStatsResponse)
+
+    const stats = this.response.payload.readInt32()
+    for (let i = 0; i < stats; i++) {
+      const statKey = this.response.payload.readString()
+      const statVal = this.response.payload.readInt64()
+      this.rawStats[statKey] = statVal
+    }
+
+    this.statistics.committedChunkId = this.rawStats["committed_chunk_id"]
+    this.statistics.firstChunkId = this.rawStats["first_chunk_id"]
+    this.statistics.lastChunkId = this.rawStats["last_chunk_id"]
+  }
+}

test/unit/stream_stats.test.ts

@@ -0,0 +1,45 @@
+import { Connection } from "../../src"
+import { expect } from "chai"
+import { Rabbit } from "../support/rabbit"
+import { randomUUID } from "crypto"
+import { expectToThrowAsync, username, password } from "../support/util"
+import { createConnection } from "../support/fake_data"
+
+describe("StreamStats", () => {
+  const rabbit = new Rabbit(username, password)
+  const testStreamName = "test-stream"
+  let connection: Connection
+  let publisherRef: string
+
+  beforeEach(async () => {
+    publisherRef = randomUUID()
+    await rabbit.createStream(testStreamName)
+    connection = await createConnection(username, password)
+  })
+
+  afterEach(async () => {
+    await connection.close()
+    await rabbit.deleteStream(testStreamName)
+  })
+
+  it("gets statistics for a stream", async () => {
+    const publisher = await connection.declarePublisher({ stream: testStreamName, publisherRef })
+    for (let i = 0; i < 5; i++) {
+      await publisher.send(Buffer.from(`test${randomUUID()}`))
+    }
+
+    const stats = await connection.streamStatsRequest(testStreamName)
+
+    expect(stats.committedChunkId).to.be.a("BigInt")
+    expect(stats.firstChunkId).to.be.a("BigInt")
+    expect(stats.lastChunkId).to.be.a("BigInt")
+  }).timeout(10000)
+
+  it("returns an error when the stream does not exist", async () => {
+    await expectToThrowAsync(
+      () => connection.streamStatsRequest("stream-does-not-exist"),
+      Error,
+      "Stream Stats command returned error with code 2 - Stream does not exist"
+    )
+  }).timeout(10000)
+})