Merge pull request #16 from coders51/2-as-a-library-user-i-want-to-create-an-exchange-without-using-the-management-ui

2# as a library user i want to create an exchange without using the management UI

Author: apietroni51

cspell.json

@@ -4,6 +4,7 @@
     "ackmode",
     "ampq",
     "amqpvalue",
+    "fanout",
     "perftest",
     "prefetch",
     "RABBITMQ",

src/exchange.ts

@@ -0,0 +1,33 @@
+export type ExchangeType = "direct" | "fanout" | "topic" | "headers"
+
+export type ExchangeOptions = {
+  arguments: Record<string, string>
+  auto_delete: boolean
+  durable: boolean
+  type: ExchangeType
+}
+
+export interface ExchangeInfo {
+  name: string
+  arguments: Record<string, string>
+  autoDelete: boolean
+  durable: boolean
+  type: ExchangeType
+}
+
+export interface Exchange {
+  getInfo: ExchangeInfo
+}
+
+export type DeletedExchangeInfo = {
+  name: string
+  deleted: boolean
+}
+
+export class AmqpExchange implements Exchange {
+  constructor(private readonly info: ExchangeInfo) {}
+
+  public get getInfo(): ExchangeInfo {
+    return this.info
+  }
+}

src/management.ts

@@ -1,3 +1,4 @@
+import { AmqpExchange, Exchange, ExchangeInfo, ExchangeOptions } from "./exchange.js"
 import { AmqpQueue, Queue, QueueOptions, QueueType } from "./queue.js"
 import {
   EventContext,
@@ -10,7 +11,12 @@ import {
   SenderOptions,
 } from "rhea"
 import { AmqpEndpoints, AmqpMethods, MessageBuilder, ME } from "./message_builder.js"
-import { CreateQueueResponseDecoder, DeleteQueueResponseDecoder } from "./response_decoder.js"
+import {
+  CreateExchangeResponseDecoder,
+  CreateQueueResponseDecoder,
+  DeleteExchangeResponseDecoder,
+  DeleteQueueResponseDecoder,
+} from "./response_decoder.js"
 
 type LinkOpenEvents = SenderEvents.senderOpen | ReceiverEvents.receiverOpen
 type LinkErrorEvents = SenderEvents.senderError | ReceiverEvents.receiverError
@@ -30,6 +36,8 @@ const MANAGEMENT_NODE_CONFIGURATION: SenderOptions | ReceiverOptions = {
 export interface Management {
   declareQueue: (queueName: string, options?: Partial<QueueOptions>) => Promise<Queue>
   deleteQueue: (queueName: string) => Promise<boolean>
+  declareExchange: (exchangeName: string, options: Partial<ExchangeOptions>) => Promise<Exchange>
+  deleteExchange: (exchangeName: string) => Promise<boolean>
   close: () => void
 }
 
@@ -154,6 +162,67 @@ export class AmqpManagement implements Management {
       this.senderLink.send(message)
     })
   }
+
+  declareExchange(exchangeName: string, options: Partial<ExchangeOptions> = {}): Promise<Exchange> {
+    const exchangeInfo: ExchangeInfo = {
+      type: options.type ?? "direct",
+      arguments: options.arguments ?? {},
+      autoDelete: options.auto_delete ?? false,
+      durable: options.durable ?? false,
+      name: exchangeName,
+    }
+    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 CreateExchangeResponseDecoder().decodeFrom(context.message, String(message.message_id))
+        if (response.status === "error") {
+          return rej(response.error)
+        }
+
+        return res(new AmqpExchange(exchangeInfo))
+      })
+
+      const message = new MessageBuilder()
+        .sendTo(`/${AmqpEndpoints.Exchanges}/${encodeURIComponent(exchangeName)}`)
+        .setReplyTo(ME)
+        .setAmqpMethod(AmqpMethods.PUT)
+        .setBody({
+          type: options.type,
+          durable: options.durable ?? false,
+          auto_delete: options.auto_delete ?? false,
+        })
+        .build()
+
+      this.senderLink.send(message)
+    })
+  }
+
+  deleteExchange(exchangeName: 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 DeleteExchangeResponseDecoder().decodeFrom(context.message, String(message.message_id))
+        if (response.status === "error") {
+          return rej(response.error)
+        }
+
+        return res(true)
+      })
+
+      const message = new MessageBuilder()
+        .sendTo(`/${AmqpEndpoints.Exchanges}/${encodeURIComponent(exchangeName)}`)
+        .setReplyTo(ME)
+        .setAmqpMethod(AmqpMethods.DELETE)
+        .build()
+      this.senderLink.send(message)
+    })
+  }
 }
 
 function buildArgumentsFrom(queueType?: QueueType, queueOptions?: Record<string, string>) {

src/message_builder.ts

@@ -8,6 +8,7 @@ export enum AmqpMethods {
 
 export enum AmqpEndpoints {
   Queues = "queues",
+  Exchanges = "exchanges",
 }
 
 export const ME = "$me"

src/response_decoder.ts

@@ -6,6 +6,19 @@ interface ResponseDecoder {
   decodeFrom: (receivedMessage: Message, sentMessageId: string) => Result<unknown, Error>
 }
 
+class VoidResponseDecoder implements ResponseDecoder {
+  decodeFrom(receivedMessage: Message, sentMessageId: string): Result<void, Error> {
+    if (isError(receivedMessage) || sentMessageId !== receivedMessage.correlation_id) {
+      return { status: "error", error: new Error(`Message Error: ${receivedMessage.subject}`) }
+    }
+
+    return {
+      status: "ok",
+      body: undefined,
+    }
+  }
+}
+
 export class CreateQueueResponseDecoder implements ResponseDecoder {
   decodeFrom(receivedMessage: Message, sentMessageId: string): Result<QueueInfo, Error> {
     if (isError(receivedMessage) || sentMessageId !== receivedMessage.correlation_id) {
@@ -45,3 +58,7 @@ export class DeleteQueueResponseDecoder implements ResponseDecoder {
     }
   }
 }
+
+export class CreateExchangeResponseDecoder extends VoidResponseDecoder {}
+
+export class DeleteExchangeResponseDecoder extends VoidResponseDecoder {}

test/e2e/management.test.ts

@@ -1,6 +1,19 @@
 import { Management } from "../../src/index.js"
 import { afterEach, beforeEach, describe, expect, test } from "vitest"
-import { createQueue, eventually, existsQueue, getQueueInfo, host, password, port, username } from "../support/util.js"
+import {
+  createQueue,
+  existsQueue,
+  eventually,
+  createExchange,
+  existsExchange,
+  deleteExchange,
+  getQueueInfo,
+  host,
+  password,
+  port,
+  username,
+  getExchangeInfo,
+} from "../support/util.js"
 import { createEnvironment, Environment } from "../../src/environment.js"
 import { Connection } from "../../src/connection.js"
 
@@ -9,6 +22,8 @@ describe("Management", () => {
   let connection: Connection
   let management: Management
 
+  const exchangeName = "test-exchange"
+
   beforeEach(async () => {
     environment = createEnvironment({
       host,
@@ -18,13 +33,15 @@ describe("Management", () => {
     })
     connection = await environment.createConnection()
     management = connection.management()
+    await deleteExchange(exchangeName)
   })
 
   afterEach(async () => {
     try {
       await management.close()
       await connection.close()
       await environment.close()
+      await deleteExchange(exchangeName)
     } catch (error) {
       console.error(error)
     }
@@ -56,4 +73,36 @@ describe("Management", () => {
       expect(await existsQueue("test-queue")).to.eql(false)
     })
   })
+
+  test("create an exchange through the management", async () => {
+    const exchange = await management.declareExchange(exchangeName, {
+      type: "headers",
+      auto_delete: true,
+      durable: false,
+    })
+
+    await eventually(async () => {
+      const exchangeInfo = await getExchangeInfo(exchange.getInfo.name)
+      expect(exchangeInfo.ok).to.eql(true)
+      expect(exchange.getInfo.name).to.eql(exchangeInfo.body.name)
+      expect(exchange.getInfo.arguments).to.eql(exchangeInfo.body.arguments)
+      expect(exchange.getInfo.autoDelete).to.eql(exchangeInfo.body.auto_delete)
+      expect(exchange.getInfo.durable).to.eql(exchangeInfo.body.durable)
+      expect(exchange.getInfo.type).to.eql(exchangeInfo.body.type)
+    })
+  })
+
+  test("delete an exchange through the management", async () => {
+    await createExchange(exchangeName)
+    await eventually(async () => {
+      expect(await existsExchange(exchangeName)).to.eql(true)
+    })
+
+    const result = await management.deleteExchange(exchangeName)
+
+    await eventually(async () => {
+      expect(await existsExchange(exchangeName)).to.eql(false)
+      expect(result).eql(true)
+    })
+  })
 })

test/support/util.ts

@@ -19,6 +19,14 @@ export type QueueInfoResponse = {
   type: string
 }
 
+export type ExchangeInfoResponse = {
+  name: string
+  arguments: Record<string, string>
+  auto_delete: boolean
+  durable: boolean
+  type: string
+}
+
 export const host = process.env.RABBITMQ_HOSTNAME ?? "localhost"
 export const port = parseInt(process.env.RABBITMQ_PORT ?? "5672")
 export const managementPort = 15672
@@ -74,6 +82,70 @@ export async function createQueue(queue: string): Promise<boolean> {
   return response.ok
 }
 
+export async function existsExchange(exchangeName: string): Promise<boolean> {
+  const response = await getExchangeInfo(exchangeName)
+
+  if (!response.ok) {
+    if (response.statusCode === 404) return false
+
+    throw new Error(`HTTPError: ${inspect(response)}`)
+  }
+
+  return response.ok
+}
+
+export async function getExchangeInfo(exchange: string): Promise<Response<ExchangeInfoResponse>> {
+  const response = await got.get<ExchangeInfoResponse>(
+    `http://${host}:${managementPort}/api/exchanges/${vhost}/${exchange}`,
+    {
+      headers: {
+        Authorization: `Basic ${Buffer.from(`${username}:${password}`).toString("base64")}`,
+      },
+      responseType: "json",
+      throwHttpErrors: false,
+    }
+  )
+
+  console.log(response.body)
+  return response
+}
+
+export async function createExchange(exchange: string): Promise<Response<unknown>> {
+  const response = await got.put(`http://${host}:${managementPort}/api/exchanges/${vhost}/${exchange}`, {
+    headers: {
+      Authorization: `Basic ${Buffer.from(`${username}:${password}`).toString("base64")}`,
+    },
+    responseType: "json",
+    throwHttpErrors: false,
+    json: {
+      type: "direct",
+      auto_delete: false,
+      durable: false,
+      internal: false,
+      arguments: {},
+    },
+  })
+
+  console.log(response.body)
+  return response
+}
+
+export async function deleteExchange(exchange: string): Promise<Response<unknown>> {
+  const response = await got.delete(`http://${host}:${managementPort}/api/exchanges/${vhost}/${exchange}`, {
+    headers: {
+      Authorization: `Basic ${Buffer.from(`${username}:${password}`).toString("base64")}`,
+    },
+    searchParams: {
+      "if-unused": true,
+    },
+    responseType: "json",
+    throwHttpErrors: false,
+  })
+
+  console.log(response.body)
+  return response
+}
+
 export async function wait(ms: number) {
   return new Promise((res) => {
     setTimeout(() => res(true), ms)