diff --git a/src/util/util/Event.ts b/src/util/util/Event.ts
index f56d6664..ce5c25d2 100644
--- a/src/util/util/Event.ts
+++ b/src/util/util/Event.ts
@@ -20,30 +20,21 @@ import { Channel } from "amqplib";
import { RabbitMQ } from "./RabbitMQ";
import EventEmitter from "events";
import { EVENT, Event } from "../interfaces";
+import { randomUUID } from "crypto";
export const events = new EventEmitter();
export async function emitEvent(payload: Omit<Event, "created_at">) {
- const id = (payload.guild_id ||
- payload.channel_id ||
- payload.user_id) as string;
+ const id = (payload.guild_id || payload.channel_id || payload.user_id) as string;
if (!id) return console.error("event doesn't contain any id", payload);
if (RabbitMQ.connection) {
- const data =
- typeof payload.data === "object"
- ? JSON.stringify(payload.data)
- : payload.data; // use rabbitmq for event transmission
+ const data = typeof payload.data === "object" ? JSON.stringify(payload.data) : payload.data; // use rabbitmq for event transmission
await RabbitMQ.channel?.assertExchange(id, "fanout", {
durable: false,
});
// assertQueue isn't needed, because a queue will automatically created if it doesn't exist
- const successful = RabbitMQ.channel?.publish(
- id,
- "",
- Buffer.from(`${data}`),
- { type: payload.event },
- );
+ const successful = RabbitMQ.channel?.publish(id, "", Buffer.from(`${data}`), { type: payload.event });
if (!successful) throw new Error("failed to send event");
} else if (process.env.EVENT_TRANSMISSION === "process") {
process.send?.({ type: "event", event: payload, id } as ProcessEvent);
@@ -79,17 +70,10 @@ export interface ProcessEvent {
id: string;
}
-export async function listenEvent(
- event: string,
- callback: (event: EventOpts) => unknown,
- opts?: ListenEventOpts,
-) {
+export async function listenEvent(event: string, callback: (event: EventOpts) => unknown, opts?: ListenEventOpts): Promise<() => Promise<void>> {
if (RabbitMQ.connection) {
const channel = opts?.channel || RabbitMQ.channel;
- if (!channel)
- throw new Error(
- "[Events] An event was sent without an associated channel",
- );
+ if (!channel) throw new Error("[Events] An event was sent without an associated channel");
return await rabbitListen(channel, event, callback, {
acknowledge: opts?.acknowledge,
});
@@ -101,9 +85,7 @@ export async function listenEvent(
const listener = (msg: ProcessEvent) => {
// eslint-disable-next-line @typescript-eslint/no-unused-expressions
- msg.type === "event" &&
- msg.id === event &&
- callback({ ...msg.event, cancel });
+ msg.type === "event" && msg.id === event && callback({ ...msg.event, cancel });
};
// TODO: assert the type is correct?
@@ -124,20 +106,17 @@ export async function listenEvent(
}
}
-async function rabbitListen(
- channel: Channel,
- id: string,
- callback: (event: EventOpts) => unknown,
- opts?: { acknowledge?: boolean },
-) {
+async function rabbitListen(channel: Channel, id: string, callback: (event: EventOpts) => unknown, opts?: { acknowledge?: boolean }): Promise<() => Promise<void>> {
await channel.assertExchange(id, "fanout", { durable: false });
const q = await channel.assertQueue("", {
exclusive: true,
autoDelete: true,
});
+ const consumerTag = randomUUID();
+
const cancel = async () => {
- await channel.cancel(q.queue);
+ await channel.cancel(consumerTag);
await channel.unbindQueue(q.queue, id, "");
};
@@ -163,6 +142,7 @@ async function rabbitListen(
},
{
noAck: !opts?.acknowledge,
+ consumerTag: consumerTag,
},
);
diff --git a/src/util/util/RabbitMQ.ts b/src/util/util/RabbitMQ.ts
index 1a61aee9..89e8e140 100644
--- a/src/util/util/RabbitMQ.ts
+++ b/src/util/util/RabbitMQ.ts
@@ -34,7 +34,24 @@ export const RabbitMQ: {
timeout: 1000 * 60,
});
console.log(`[RabbitMQ] connected`);
+
+ // log connection errors
+ this.connection.on("error", (err) => {
+ console.error("[RabbitMQ] Connection Error:", err);
+ });
+
+ this.connection.on("close", () => {
+ console.error("[RabbitMQ] connection closed");
+ // TODO: Add reconnection logic here if the connection crashes??
+ // will be a pain since we will have to reconstruct entire state
+ });
+
this.channel = await this.connection.createChannel();
console.log(`[RabbitMQ] channel created`);
+
+ // log channel errors
+ this.channel.on("error", (err) => {
+ console.error("[RabbitMQ] Channel Error:", err);
+ });
},
};
|
