dandri/server
1
0
Fork 0
mirror of https://github.com/spacebarchat/server.git synced 2026-08-26 02:39:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move publishEvent to RabbitMQ.ts

Browse Source
This commit is contained in:
Rory&
2026-05-25 09:21:25 +02:00
committed by Rory&
parent 31961be027
commit 7a5d0d0d76
2 changed files with 28 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/util/util/ipc/Event.ts
+1 -28
View File
@@ -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!");
src/util/util/ipc/RabbitMQ.ts
+27 -1
View File
@@ -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>> {