diff --git a/src/util/util/ipc/Event.ts b/src/util/util/ipc/Event.ts
index 3eb052b9..8a2c55d8 100644
--- a/src/util/util/ipc/Event.ts
+++ b/src/util/util/ipc/Event.ts
@@ -40,34 +40,7 @@ export async function emitEvent(payload: Omit<Event, "created_at">) {
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 publishEvent = async (retryCount = 0): Promise<void> => {
- const channel = await RabbitMQ.getSafeChannel();
- try {
- await channel.assertExchange(id, "fanout", {
- durable: false,
- });
-
- // assertQueue isn't needed, because a queue will automatically created if it doesn't exist
- const successful = channel.publish(id, "", Buffer.from(`${data}`), { type: payload.event });
- if (!successful) throw new Error("failed to send event");
- } catch (e) {
- // Check if this is a channel closed error and if we should retry
- const errorMessage = e instanceof Error ? e.message : String(e);
- const isChannelError = errorMessage.includes("Channel closed") || errorMessage.includes("IllegalOperationError") || errorMessage.includes("RESOURCE_ERROR");
-
- if (isChannelError && retryCount < 1) {
- console.log("[RabbitMQ] Channel error detected, retrying with new channel...");
- // Force the cached channel to be discarded by calling getSafeChannel which will create a new one
- return publishEvent(retryCount + 1);
- }
-
- console.log("[RabbitMQ] ", e);
- }
- };
-
- await publishEvent();
+ await RabbitMQ.publishEvent(id, payload);
} else if (process.env.EVENT_TRANSMISSION === "unix" && process.env.EVENT_SOCKET_PATH) {
if (!unixSocketWriter) {
console.error("[Event] Unix socket writer not initialized, cannot emit event!");
diff --git a/src/util/util/ipc/RabbitMQ.ts b/src/util/util/ipc/RabbitMQ.ts
index 47d1660d..e3a1ca4f 100644
--- a/src/util/util/ipc/RabbitMQ.ts
+++ b/src/util/util/ipc/RabbitMQ.ts
@@ -19,7 +19,7 @@
import { randomUUID } from "node:crypto";
import EventEmitter from "node:events";
import amqp, { Channel, ChannelModel } from "amqplib";
-import { EVENT } from "../../interfaces";
+import { Event, EVENT } from "../../interfaces";
import { Config } from "../Config";
import type { EventOpts } from "./Event";
@@ -173,6 +173,32 @@ export class RabbitMQ {
static isConnected(): boolean {
return this.connection !== null && !this.isReconnecting;
}
+
+ static async publishEvent(id: string, payload: Omit<Event, "created_at">, retryCount = 0): Promise<void> {
+ const data = typeof payload.data === "object" ? JSON.stringify(payload.data) : payload.data; // use rabbitmq for event transmission
+ const channel = await RabbitMQ.getSafeChannel();
+ try {
+ await channel.assertExchange(id, "fanout", {
+ durable: false,
+ });
+
+ // assertQueue isn't needed, because a queue will automatically created if it doesn't exist
+ const successful = channel.publish(id, "", Buffer.from(`${data}`), { type: payload.event });
+ if (!successful) throw new Error("failed to send event");
+ } catch (e) {
+ // Check if this is a channel closed error and if we should retry
+ const errorMessage = e instanceof Error ? e.message : String(e);
+ const isChannelError = errorMessage.includes("Channel closed") || errorMessage.includes("IllegalOperationError") || errorMessage.includes("RESOURCE_ERROR");
+
+ if (isChannelError && retryCount < 1) {
+ console.log("[RabbitMQ] Channel error detected, retrying with new channel...");
+ // Force the cached channel to be discarded by calling getSafeChannel which will create a new one
+ return RabbitMQ.publishEvent(id, payload, retryCount + 1);
+ }
+
+ console.log("[RabbitMQ] ", e);
+ }
+ }
}
export async function rabbitListen(channel: Channel, id: string, callback: (event: EventOpts) => unknown, opts?: { acknowledge?: boolean }): Promise<() => Promise<void>> {
|
