added response

Author: apietroni51

src/exchange.ts

@@ -1,23 +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: string
 }
 
-export class AmqpExchangeInfo implements ExchangeInfo {
-  private exchangeName: string
+export interface Exchange {
+  getInfo: ExchangeInfo
+}
 
-  constructor(params: { name: string }) {
-    this.exchangeName = params.name
-  }
+export type DeletedExchangeInfo = {
+  name: string
+  deleted: boolean
+}
+
+export class AmqpExchange implements Exchange {
+  constructor(private readonly info: ExchangeInfo) {}
 
-  public get name(): string {
-    return this.exchangeName
+  public get getInfo(): ExchangeInfo {
+    return this.info
   }
 }

src/management.ts

@@ -1,4 +1,4 @@
-import { AmqpExchangeInfo, ExchangeInfo, ExchangeOptions } from "./exchange.js"
+import { AmqpExchange, Exchange, ExchangeInfo, ExchangeOptions } from "./exchange.js"
 import { AmqpQueue, Queue, QueueOptions, QueueType } from "./queue.js"
 import {
   EventContext,
@@ -11,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
@@ -31,8 +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>) => ExchangeInfo
-  deleteExchange: (exchangeName: string) => void
+  declareExchange: (exchangeName: string, options: Partial<ExchangeOptions>) => Promise<Exchange>
+  deleteExchange: (exchangeName: string) => Promise<boolean>
   close: () => void
 }
 
@@ -158,28 +163,64 @@ export class AmqpManagement implements Management {
     })
   }
 
-  declareExchange(exchangeName: string, options: Partial<ExchangeOptions> = {}): ExchangeInfo {
-    // decode the response
-    // create queueInfo
+  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"))
+        }
 
-    this.senderLink.send({
-      message_id: generate_uuid(),
-      to: `/exchanges/${encodeURIComponent(exchangeName)}`,
-      reply_to: "$me",
-      subject: "PUT",
-      body: options,
-    })
+        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()
 
-    return new AmqpExchangeInfo({ name: exchangeName })
+      this.senderLink.send(message)
+    })
   }
 
-  deleteExchange(exchangeName: string) {
-    this.senderLink.send({
-      message_id: generate_uuid(),
-      to: `/exchanges/${encodeURIComponent(exchangeName)}`,
-      reply_to: "$me",
-      subject: "DELETE",
-      body: null,
+  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)
     })
   }
 }

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

@@ -45,3 +45,29 @@ export class DeleteQueueResponseDecoder implements ResponseDecoder {
     }
   }
 }
+
+export class CreateExchangeResponseDecoder 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 DeleteExchangeResponseDecoder 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,
+    }
+  }
+}

test/e2e/management.test.ts

@@ -12,6 +12,7 @@ import {
   password,
   port,
   username,
+  getExchangeInfo,
 } from "../support/util.js"
 import { createEnvironment, Environment } from "../../src/environment.js"
 import { Connection } from "../../src/connection.js"
@@ -74,15 +75,20 @@ describe("Management", () => {
   })
 
   test("create an exchange through the management", async () => {
-    const exchangeInfo = management.declareExchange(exchangeName, {
+    const exchange = await management.declareExchange(exchangeName, {
       type: "headers",
       auto_delete: true,
       durable: false,
     })
 
-    expect(exchangeInfo.name).to.eql(exchangeName)
     await eventually(async () => {
-      expect(await existsExchange(exchangeInfo.name)).to.eql(true)
+      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)
     })
   })
 
@@ -92,10 +98,11 @@ describe("Management", () => {
       expect(await existsExchange(exchangeName)).to.eql(true)
     })
 
-    management.deleteExchange(exchangeName)
+    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
@@ -86,8 +94,8 @@ export async function existsExchange(exchangeName: string): Promise<boolean> {
   return response.ok
 }
 
-async function getExchangeInfo(exchange: string): Promise<Response<QueueInfoResponse>> {
-  const response = await got.get<QueueInfoResponse>(
+export async function getExchangeInfo(exchange: string): Promise<Response<ExchangeInfoResponse>> {
+  const response = await got.get<ExchangeInfoResponse>(
     `http://${host}:${managementPort}/api/exchanges/${vhost}/${exchange}`,
     {
       headers: {