feat: add create and delete queue

Author: magne

src/management.ts

@@ -1,6 +1,6 @@
-import { AmqpQueueInfo, QueueInfo } from "./queue.js"
+import { AmqpQueue, Queue, QueueOptions, QueueType } from "./queue.js"
 import {
-  generate_uuid,
+  EventContext,
   Receiver,
   ReceiverEvents,
   ReceiverOptions,
@@ -9,11 +9,8 @@ import {
   SenderEvents,
   SenderOptions,
 } from "rhea"
-
-type QueueOptions = {
-  exclusive: boolean
-  auto_delete: boolean
-}
+import { AmqpEndpoints, AmqpMethods, MessageBuilder, ME } from "./message_builder.js"
+import { CreateQueueResponseDecoder, DeleteQueueResponseDecoder } from "./response_decoder.js"
 
 type LinkOpenEvents = SenderEvents.senderOpen | ReceiverEvents.receiverOpen
 type LinkErrorEvents = SenderEvents.senderError | ReceiverEvents.receiverError
@@ -31,27 +28,26 @@ const MANAGEMENT_NODE_CONFIGURATION: SenderOptions | ReceiverOptions = {
 }
 
 export interface Management {
-  declareQueue: (queueName: string, options: Partial<QueueOptions>) => QueueInfo
+  declareQueue: (queueName: string, options?: Partial<QueueOptions>) => Promise<Queue>
+  deleteQueue: (queueName: string) => Promise<boolean>
   close: () => void
 }
 
 export class AmqpManagement implements Management {
   static async create(connection: RheaConnection): Promise<AmqpManagement> {
     const senderLink = await AmqpManagement.openSender(connection)
     const receiverLink = await AmqpManagement.openReceiver(connection)
-    return new AmqpManagement(senderLink, receiverLink)
+    return new AmqpManagement(connection, senderLink, receiverLink)
   }
 
   constructor(
-    // private readonly connection: RheaConnection,
+    private readonly connection: RheaConnection,
     private senderLink: Sender,
     private receiverLink: Receiver
   ) {
     console.log(this.receiverLink.is_open())
   }
 
-  async close() {}
-
   private static async openReceiver(connection: RheaConnection): Promise<Receiver> {
     return AmqpManagement.openLink<Receiver>(
       connection,
@@ -90,18 +86,76 @@ export class AmqpManagement implements Management {
     })
   }
 
-  declareQueue(queueName: string, options: Partial<QueueOptions> = {}): QueueInfo {
-    // decode the response
-    // create queueInfo
+  close(): void {
+    if (this.connection.is_closed()) return
+
+    this.closeSender()
+    this.closeReceiver()
+  }
+
+  private closeSender(): void {
+    this.senderLink.close()
+  }
+
+  private closeReceiver(): void {
+    this.senderLink.close()
+  }
+
+  async declareQueue(queueName: string, options: Partial<QueueOptions> = {}): 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 CreateQueueResponseDecoder().decodeFrom(context.message, String(message.message_id))
+        if (response.status === "error") {
+          return rej(response.error)
+        }
 
-    this.senderLink.send({
-      message_id: generate_uuid(),
-      to: `/queues/${encodeURIComponent(queueName)}`,
-      reply_to: "$me",
-      subject: "PUT",
-      body: options,
+        return res(new AmqpQueue(response.body))
+      })
+
+      const message = new MessageBuilder()
+        .sendTo(`/${AmqpEndpoints.Queues}/${encodeURIComponent(queueName)}`)
+        .setReplyTo(ME)
+        .setAmqpMethod(AmqpMethods.PUT)
+        .setBody({
+          exclusive: options.exclusive ?? false,
+          durable: options.durable ?? false,
+          auto_delete: options.autoDelete ?? false,
+          arguments: buildArgumentsFrom(options.type, options.arguments),
+        })
+        .build()
+      this.senderLink.send(message)
     })
+  }
+
+  async deleteQueue(queueName: string): Promise<boolean> {
+    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 DeleteQueueResponseDecoder().decodeFrom(context.message, String(message.message_id))
+        if (response.status === "error") {
+          return rej(response.error)
+        }
 
-    return new AmqpQueueInfo({ name: queueName })
+        return res(true)
+      })
+
+      const message = new MessageBuilder()
+        .sendTo(`/${AmqpEndpoints.Queues}/${encodeURIComponent(queueName)}`)
+        .setReplyTo(ME)
+        .setAmqpMethod(AmqpMethods.DELETE)
+        .build()
+      this.senderLink.send(message)
+    })
   }
 }
+
+function buildArgumentsFrom(queueType?: QueueType, queueOptions?: Record<string, string>) {
+  return { ...(queueOptions ?? {}), ...(queueType ? { "x-queue-type": queueType } : {}) }
+}

src/message_builder.ts

@@ -0,0 +1,58 @@
+import { generate_uuid, Message } from "rhea"
+
+export enum AmqpMethods {
+  PUT = "PUT",
+  DELETE = "DELETE",
+  GET = "GET",
+}
+
+export enum AmqpEndpoints {
+  Queues = "queues",
+}
+
+export const ME = "$me"
+
+export class MessageBuilder {
+  private messageId: string = generate_uuid()
+  private to: string = ""
+  private replyTo: string = ME
+  private method: AmqpMethods = AmqpMethods.GET
+  private body: unknown
+
+  constructor() {}
+
+  setMessageId(id: string) {
+    this.messageId = id
+    return this
+  }
+
+  sendTo(to: string) {
+    this.to = to
+    return this
+  }
+
+  setReplyTo(replyTo: string) {
+    this.replyTo = replyTo
+    return this
+  }
+
+  setAmqpMethod(method: AmqpMethods) {
+    this.method = method
+    return this
+  }
+
+  setBody(body: unknown) {
+    this.body = body
+    return this
+  }
+
+  build(): Message {
+    return {
+      message_id: this.messageId,
+      to: this.to,
+      reply_to: this.replyTo,
+      subject: this.method,
+      body: this.body,
+    }
+  }
+}

src/queue.ts

@@ -1,15 +1,39 @@
-export interface QueueInfo {
+export type QueueType = "classic" | "stream" | "quorum"
+
+export type QueueOptions = {
+  type: QueueType
+  exclusive: boolean
+  autoDelete: boolean
+  durable: boolean
+  arguments: Record<string, string>
+}
+
+export type QueueInfo = {
+  name: string
+  durable: boolean
+  autoDelete: boolean
+  exclusive: boolean
+  type: QueueType
+  leader: string
+  replicas: string[]
+  messageCount: number
+  consumerCount: number
+  arguments: Record<string, string>
+}
+
+export type DeletedQueueInfo = {
   name: string
+  deleted: boolean
 }
 
-export class AmqpQueueInfo implements QueueInfo {
-  private queueName: string
+export interface Queue {
+  getInfo: QueueInfo
+}
 
-  constructor(params: { name: string }) {
-    this.queueName = params.name
-  }
+export class AmqpQueue implements Queue {
+  constructor(private readonly info: QueueInfo) {}
 
-  public get name(): string {
-    return this.queueName
+  public get getInfo(): QueueInfo {
+    return this.info
   }
 }

src/response_decoder.ts

@@ -0,0 +1,47 @@
+import { Message } from "rhea"
+import { AUTO_DELETE, DURABLE, EXCLUSIVE, isError, queueTypeFromString, Result } from "./utils.js"
+import { DeletedQueueInfo, QueueInfo } from "./queue.js"
+
+interface ResponseDecoder {
+  decodeFrom: (receivedMessage: Message, sentMessageId: string) => unknown
+}
+
+export class CreateQueueResponseDecoder implements ResponseDecoder {
+  decodeFrom(receivedMessage: Message, sentMessageId: string): Result<QueueInfo, Error> {
+    if (isError(receivedMessage) || sentMessageId !== receivedMessage.correlation_id) {
+      return { status: "error", error: new Error(`Message Error: ${receivedMessage.subject}`) }
+    }
+
+    return {
+      status: "ok",
+      body: {
+        name: receivedMessage.body.name,
+        durable: receivedMessage.body.durable === DURABLE,
+        autoDelete: receivedMessage.body.auto_delete === AUTO_DELETE,
+        exclusive: receivedMessage.body.exclusive === EXCLUSIVE,
+        type: queueTypeFromString(receivedMessage.body.type),
+        arguments: receivedMessage.body.arguments ?? {},
+        leader: receivedMessage.body.leader,
+        replicas: receivedMessage.body.replicas,
+        messageCount: receivedMessage.body.message_count,
+        consumerCount: receivedMessage.body.consumer_count,
+      },
+    }
+  }
+}
+
+export class DeleteQueueResponseDecoder implements ResponseDecoder {
+  decodeFrom(receivedMessage: Message, sentMessageId: string): Result<DeletedQueueInfo, Error> {
+    if (isError(receivedMessage) || sentMessageId !== receivedMessage.correlation_id) {
+      return { status: "error", error: new Error(`Message Error: ${receivedMessage.subject}`) }
+    }
+
+    return {
+      status: "ok",
+      body: {
+        name: receivedMessage.body.name,
+        deleted: true,
+      },
+    }
+  }
+}

src/utils.ts

@@ -0,0 +1,48 @@
+import { Message } from "rhea"
+import { QueueType } from "./queue.js"
+
+export enum AmqpResponseCodes {
+  OK = "200",
+  CREATED = "201",
+  NO_CONTENT = "204",
+  BAD_REQUEST = "400",
+  NOT_FOUND = "404",
+  CONFLICT = "409",
+}
+
+export const DURABLE = 1
+export const AUTO_DELETE = 1
+export const EXCLUSIVE = 1
+
+export type Result<T, K> = OkResult<T> | ErrorResult<K>
+
+type OkResult<T> = {
+  status: "ok"
+  body: T
+}
+
+type ErrorResult<K> = {
+  status: "error"
+  error: K
+}
+
+export function isError(message: Message): boolean {
+  return (
+    message.subject === AmqpResponseCodes.BAD_REQUEST ||
+    message.subject === AmqpResponseCodes.NOT_FOUND ||
+    message.subject === AmqpResponseCodes.CONFLICT
+  )
+}
+
+export function queueTypeFromString(queueType: string): QueueType {
+  switch (queueType) {
+    case "classic":
+      return "classic"
+    case "quorum":
+      return "quorum"
+    case "stream":
+      return "stream"
+    default:
+      throw new Error(`Unsupported queue type: ${queueType}`)
+  }
+}

test/e2e/management.test.ts

@@ -1,6 +1,6 @@
 import { Management } from "../../src/index.js"
 import { afterEach, beforeEach, describe, expect, test } from "vitest"
-import { eventually, existsQueue, host, password, port, username } from "../support/util.js"
+import { createQueue, eventually, existsQueue, getQueueInfo, host, password, port, username } from "../support/util.js"
 import { createEnvironment, Environment } from "../../src/environment.js"
 import { Connection } from "../../src/connection.js"
 
@@ -21,17 +21,39 @@ describe("Management", () => {
   })
 
   afterEach(async () => {
-    await management.close()
-    await connection.close()
-    await environment.close()
+    try {
+      await management.close()
+      await connection.close()
+      await environment.close()
+    } catch (error) {
+      console.error(error)
+    }
   })
 
   test("create a queue through the management", async () => {
-    const queueInfo = management.declareQueue("test-queue", { exclusive: true, auto_delete: false })
+    const queue = await management.declareQueue("test-queue")
 
-    expect(queueInfo.name).to.eql("test-queue")
     await eventually(async () => {
-      expect(await existsQueue(queueInfo.name)).to.eql(true)
+      const queueInfo = await getQueueInfo(queue.getInfo.name)
+      expect(queueInfo.ok).to.eql(true)
+      expect(queue.getInfo.arguments).to.eql(queueInfo.body.arguments)
+      expect(queue.getInfo.autoDelete).to.eql(queueInfo.body.auto_delete)
+      expect(queue.getInfo.durable).to.eql(queueInfo.body.durable)
+      expect(queue.getInfo.exclusive).to.eql(queueInfo.body.exclusive)
+      expect(queue.getInfo.consumerCount).to.eql(queueInfo.body.consumers)
+      expect(queue.getInfo.messageCount).to.eql(queueInfo.body.messages)
+      expect(queue.getInfo.type).to.eql(queueInfo.body.type)
+      expect(queue.getInfo.leader).to.eql(queueInfo.body.node)
+    })
+  })
+
+  test("delete a queue through the management", async () => {
+    await createQueue("test-queue")
+
+    await management.deleteQueue("test-queue")
+
+    await eventually(async () => {
+      expect(await existsQueue("test-queue")).to.eql(false)
     })
   })
 })