Repository navigation
Expand file tree
/
Copy pathevents.ts
More file actions
132 lines (106 loc) · 3.24 KB
/
Copy pathevents.ts
File metadata and controls
132 lines (106 loc) · 3.24 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import EventEmitter from "node:events";
import * as rabbit from "rabbitmq-stream-js-client";
import type { BaseModel } from "../entity/basemodel.js";
import type { GATEWAY_EVENT } from "../gateway/util/validation/send.js";
import { config } from "./config.js";
import { createLogger } from "./log.js";
import { getTableName } from "./entity/util.js";
const Log = createLogger("events");
const events = new EventEmitter();
// TODO: move rabbitmq handlers to own files
let startedRabbitmq = false;
let client: rabbit.Client | null = null;
let publisher: rabbit.Publisher | null = null;
export const initRabbitMQ = async (consume: boolean) => {
if (startedRabbitmq) return;
if (!config().rabbitmq.enabled) return;
startedRabbitmq = true;
const url = config().rabbitmq.url;
const STREAM_NAME = "gateway";
Log.msg(`Connecting to ${url.href}`);
client = await rabbit.connect({
hostname: url.hostname,
port: Number.parseInt(url.port, 10),
username: url.username,
password: url.password,
vhost: url.pathname,
heartbeat: 0,
});
Log.msg("Connected to RabbitMQ");
await client.createStream({
stream: STREAM_NAME,
arguments: {
"max-age": "1h",
},
});
Log.verbose(`Created stream ${STREAM_NAME}`);
publisher = await client.declarePublisher({
stream: STREAM_NAME,
publisherRef: "gateway",
});
Log.verbose("Created publisher");
if (consume) {
await client.declareConsumer(
{ stream: STREAM_NAME, offset: rabbit.Offset.next() },
(message) => {
const parsed = JSON.parse(message.content.toString());
const target = message.messageProperties?.to;
if (!target) {
Log.warn("Received rabbitmq message without `to` field");
return;
}
events.emit(target, parsed);
},
);
Log.verbose("Created consumer");
}
};
export const closeRabbitMQ = async () => {
await client?.close();
};
/**
* Emit an event via gateway to users listening to targets
* @param targets IDs of relevant objects
* @param payload Gateway payload to send to clients
* @param isPrivate For user targets - whether this event should only been delivered to the target user and not to other users listening to their ID
*/
export const emitGatewayEvent = (
targets: BaseModel | BaseModel[],
payload: GATEWAY_EVENT,
isPrivate = false,
) => {
if (!Array.isArray(targets)) targets = [targets];
const targetIds = targets.map(makeGatewayTarget);
if (publisher) {
// rabbitmq emit
for (const target of targetIds) {
publisher
.send(Buffer.from(JSON.stringify(payload)), {
messageProperties: { to: target },
})
.catch(() => {
Log.error("Failed to emit rabbitmq event?");
});
}
} else {
// normal event emit
for (const target of targetIds) {
events.emit(target + (isPrivate ? "-only" : ""), payload);
}
}
};
export const makeGatewayTarget = (target: BaseModel) => {
return `${getTableName(target)}:${target.id}`;
};
export const listenGatewayEvent = (
target: BaseModel | string,
callback: (payload: GATEWAY_EVENT) => unknown,
) => {
const id = typeof target === "string" ? target : makeGatewayTarget(target);
events.setMaxListeners(events.getMaxListeners() + 1);
events.addListener(id, callback);
return () => {
events.removeListener(id, callback);
events.setMaxListeners(events.getMaxListeners() - 1);
};
};