feat: add get queue info method (#43)

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

Author: l4mby

src/management.ts

@@ -18,6 +18,7 @@ import {
   DeleteBindingResponseDecoder,
   DeleteExchangeResponseDecoder,
   DeleteQueueResponseDecoder,
+  GetQueueInfoResponseDecoder,
 } from "./response_decoder.js"
 import { AmqpBinding, Binding, BindingInfo, BindingOptions } from "./binding.js"
 import { randomUUID } from "crypto"
@@ -35,6 +36,7 @@ const MANAGEMENT_NODE_CONFIGURATION: SenderOptions | ReceiverOptions = {
 export interface Management {
   declareQueue: (queueName: string, options?: Partial<QueueOptions>) => Promise<Queue>
   deleteQueue: (queueName: string) => Promise<boolean>
+  getQueueInfo: (queueName: string) => Promise<Queue>
   declareExchange: (exchangeName: string, options?: Partial<ExchangeOptions>) => Promise<Exchange>
   deleteExchange: (exchangeName: string) => Promise<boolean>
   bind: (key: string, options: BindingOptions) => Promise<Binding>
@@ -144,6 +146,30 @@ export class AmqpManagement implements Management {
     })
   }
 
+  async getQueueInfo(queueName: string): Promise<Queue> {
+    return new Promise((res, rej) => {
+      this.receiverLink.once(ReceiverEvents.message, (context: EventContext) => {
+        if (!context.message) {
+          return rej(new Error("Receiver has not received any message"))
+        }
+
+        const response = new GetQueueInfoResponseDecoder().decodeFrom(context.message, String(message.message_id))
+        if (response.status === "error") {
+          return rej(response.error)
+        }
+
+        return res(new AmqpQueue(response.body))
+      })
+
+      const message = new LinkMessageBuilder()
+        .sendTo(`/${AmqpEndpoints.Queues}/${encodeURIComponent(queueName)}`)
+        .setReplyTo(ME)
+        .setAmqpMethod(AmqpMethods.GET)
+        .build()
+      this.senderLink.send(message)
+    })
+  }
+
   async declareExchange(exchangeName: string, options: Partial<ExchangeOptions> = {}): Promise<Exchange> {
     const exchangeInfo: ExchangeInfo = {
       type: options.type ?? "direct",

src/response_decoder.ts

@@ -30,6 +30,8 @@ export class CreateQueueResponseDecoder implements ResponseDecoder {
   }
 }
 
+export class GetQueueInfoResponseDecoder extends CreateQueueResponseDecoder {}
+
 export class DeleteQueueResponseDecoder implements ResponseDecoder {
   decodeFrom(receivedMessage: Message, sentMessageId: string): Result<DeletedQueueInfo, Error> {
     if (isError(receivedMessage) || sentMessageId !== receivedMessage.correlation_id) {

test/e2e/management.test.ts

@@ -78,6 +78,24 @@ describe("Management", () => {
     })
   })
 
+  test("get info of a queue through the management", async () => {
+    await createQueue(queueName)
+
+    const result = await management.getQueueInfo(queueName)
+
+    await eventually(async () => {
+      const queueInfo = await getQueueInfo(queueName)
+      expect(result.getInfo.arguments).to.eql(queueInfo.body.arguments)
+      expect(result.getInfo.autoDelete).to.eql(queueInfo.body.auto_delete)
+      expect(result.getInfo.durable).to.eql(queueInfo.body.durable)
+      expect(result.getInfo.exclusive).to.eql(queueInfo.body.exclusive)
+      expect(result.getInfo.consumerCount).to.eql(queueInfo.body.consumers)
+      expect(result.getInfo.messageCount).to.eql(queueInfo.body.messages)
+      expect(result.getInfo.type).to.eql(queueInfo.body.type)
+      expect(result.getInfo.leader).to.eql(queueInfo.body.node)
+    })
+  })
+
   test("create an exchange through the management", async () => {
     const exchange = await management.declareExchange(exchangeName, {
       type: "headers",