diff --git a/src/util/util/ipc/Event.ts b/src/util/util/ipc/Event.ts
index 4899b5d8..3eb052b9 100644
--- a/src/util/util/ipc/Event.ts
+++ b/src/util/util/ipc/Event.ts
@@ -16,19 +16,18 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
-import { Channel } from "amqplib";
-import { randomUUID } from "node:crypto";
import EventEmitter from "node:events";
import path from "node:path";
-import { RabbitMQ } from "./RabbitMQ";
-import { EVENT, Event } from "../../interfaces";
+import { Channel } from "amqplib";
+import { yellow } from "picocolors";
+import { Event } from "../../interfaces";
import { Config } from "../Config";
+import { rabbitListen, RabbitMQ } from "./RabbitMQ";
import { BaseEventListener } from "./listener/BaseEventListener";
import { BaseEventWriter } from "./writer/BaseEventWriter";
import { UnixSocketWriter } from "./writer/UnixSocketWriter";
import { UnixSocketListener } from "./listener/UnixSocketListener";
import { RabbitMqSingleListener } from "./listener/RabbitMqSingleListener";
-import { yellow } from "picocolors";
export const events = new EventEmitter();
let listener: BaseEventListener | null = null;
@@ -186,51 +185,3 @@ export async function listenEvent(event: string, callback: (event: EventOpts) =>
return cancel;
}
}
-
-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,
- messageTtl: 5000,
- });
-
- const consumerTag = randomUUID();
-
- const cancel = async () => {
- try {
- await channel.unbindQueue(q.queue, id, "");
- await channel.cancel(consumerTag);
- } catch (e) {
- console.log("[RabbitMQ] Error while cancelling channel (may be expected):", e instanceof Error ? e.message : e);
- }
- };
-
- await channel.bindQueue(q.queue, id, "");
- await channel.consume(
- q.queue,
- (opts) => {
- if (!opts) return;
-
- const data = JSON.parse(opts.content.toString());
- const event = opts.properties.type as EVENT;
-
- callback({
- event,
- data,
- acknowledge() {
- channel.ack(opts);
- },
- channel,
- cancel,
- });
- // rabbitCh.ack(opts);
- },
- {
- noAck: !opts?.acknowledge,
- consumerTag: consumerTag,
- },
- );
-
- return cancel;
-}
diff --git a/src/util/util/ipc/RabbitMQ.ts b/src/util/util/ipc/RabbitMQ.ts
index fe6e6d1b..47d1660d 100644
--- a/src/util/util/ipc/RabbitMQ.ts
+++ b/src/util/util/ipc/RabbitMQ.ts
@@ -16,9 +16,12 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
+import { randomUUID } from "node:crypto";
import EventEmitter from "node:events";
import amqp, { Channel, ChannelModel } from "amqplib";
+import { EVENT } from "../../interfaces";
import { Config } from "../Config";
+import type { EventOpts } from "./Event";
export class RabbitMQ {
public static connection: ChannelModel | null = null;
@@ -171,3 +174,51 @@ export class RabbitMQ {
return this.connection !== null && !this.isReconnecting;
}
}
+
+export 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,
+ messageTtl: 5000,
+ });
+
+ const consumerTag = randomUUID();
+
+ const cancel = async () => {
+ try {
+ await channel.unbindQueue(q.queue, id, "");
+ await channel.cancel(consumerTag);
+ } catch (e) {
+ console.log("[RabbitMQ] Error while cancelling channel (may be expected):", e instanceof Error ? e.message : e);
+ }
+ };
+
+ await channel.bindQueue(q.queue, id, "");
+ await channel.consume(
+ q.queue,
+ (opts) => {
+ if (!opts) return;
+
+ const data = JSON.parse(opts.content.toString());
+ const event = opts.properties.type as EVENT;
+
+ callback({
+ event,
+ data,
+ acknowledge() {
+ channel.ack(opts);
+ },
+ channel,
+ cancel,
+ });
+ // rabbitCh.ack(opts);
+ },
+ {
+ noAck: !opts?.acknowledge,
+ consumerTag: consumerTag,
+ },
+ );
+
+ return cancel;
+}
|
