Merge branch 'fosscord:master' into master

bundle/scripts/benchmark/connections.js

@@ -0,0 +1,58 @@
+const cluster = require("cluster");
+const WebSocket = require("ws");
+const endpoint = process.env.GATEWAY || "ws://localhost:3001";
+const connections = Number(process.env.CONNECTIONS) || 50;
+const threads = Number(process.env.THREADS) || require("os").cpus().length || 1;
+const token = process.env.TOKEN;
+
+if (!token) {
+	console.error("TOKEN env var missing");
+	process.exit();
+}
+
+if (cluster.isMaster) {
+	for (let i = 0; i < threads; i++) {
+		cluster.fork();
+	}
+
+	cluster.on("exit", (worker, code, signal) => {
+		console.log(`worker ${worker.process.pid} died`);
+	});
+} else {
+	for (let i = 0; i < connections; i++) {
+		connect();
+	}
+}
+
+function connect() {
+	const client = new WebSocket(endpoint);
+	client.on("message", (data) => {
+		data = JSON.parse(data);
+
+		switch (data.op) {
+			case 10:
+				client.interval = setInterval(() => {
+					client.send(JSON.stringify({ op: 1 }));
+				}, data.d.heartbeat_interval);
+
+				client.send(
+					JSON.stringify({
+						op: 2,
+						d: {
+							token,
+							properties: {},
+						},
+					})
+				);
+
+				break;
+		}
+	});
+	client.once("close", (code, reason) => {
+		clearInterval(client.interval);
+		connect();
+	});
+	client.on("error", (err) => {
+		// console.log(err);
+	});
+}

bundle/scripts/benchmark/index.js

@@ -0,0 +1,4 @@
+require("dotenv").config();
+
+require("./connections");
+require("./messages");

bundle/scripts/benchmark/messages.js

@@ -0,0 +1 @@
+// TODO

bundle/scripts/build.js

@@ -36,8 +36,8 @@ console.log(
 	execSync(
 		"node \"" +
 			path.join(__dirname, "..", "node_modules", "typescript", "lib", "tsc.js") +
-			"\" -p " +
+			"\" -p \"" +
-			path.join(__dirname, ".."),
+			path.join(__dirname, "..") + "\"",
 		{
 			cwd: path.join(__dirname, ".."),
 			shell: true,

bundle/src/stats.ts

@@ -31,5 +31,6 @@ export function initStats() {
 				process.memoryUsage().rss / 1024 / 1024
 			)}mb/${memory.totalMemMb.toFixed(0)}mb ${networkUsage}`
 		);
-	}, 1000 * 10);
+		// TODO: node-os-utils might have a memory leak, more investigation needed
+	}, 1000 * 60 * 5);
 }

gateway/src/Server.ts

@@ -32,7 +32,6 @@ export class Server {
 		}
 
 		this.server.on("upgrade", (request, socket, head) => {
-			console.log("socket requests upgrade", request.url);
 			// @ts-ignore
 			this.ws.handleUpgrade(request, socket, head, (socket) => {
 				this.ws.emit("connection", socket, request);

gateway/src/events/Close.ts

@@ -1,10 +1,13 @@
 import { WebSocket } from "@fosscord/gateway";
-import { Message } from "./Message";
 import { Session } from "@fosscord/util";
 
 export async function Close(this: WebSocket, code: number, reason: string) {
 	console.log("[WebSocket] closed", code, reason);
 	if (this.session_id) await Session.delete({ session_id: this.session_id });
-	// @ts-ignore
+	if (this.heartbeatTimeout) clearTimeout(this.heartbeatTimeout);
-	this.off("message", Message);
+	if (this.readyTimeout) clearTimeout(this.readyTimeout);
+
+	this.deflate?.close();
+
+	this.removeAllListeners();
 }

gateway/src/events/Connection.ts

@@ -28,6 +28,7 @@ export async function Connection(
 		socket.on("close", Close);
 		// @ts-ignore
 		socket.on("message", Message);
+		console.log(`[Gateway] Connections: ${this.clients.size}`);
 
 		const { searchParams } = new URL(`http://localhost${request.url}`);
 		// @ts-ignore

gateway/src/events/Message.ts

@@ -37,8 +37,6 @@ export async function Message(this: WebSocket, buffer: WS.Data) {
 		return;
 	}
 
-	console.log("[Gateway] Opcode " + OPCODES[data.op]);
-
 	try {
 		return await OPCodeHandler.call(this, data);
 	} catch (error) {

gateway/src/opcodes/Identify.ts

@@ -214,8 +214,6 @@ export async function onIdentify(this: WebSocket, data: Payload) {
 		// application // TODO for applications
 	};
 
-	console.log("Send ready");
-
 	// TODO: send real proper data structure
 	await Send(this, {
 		op: OPCODES.Dispatch,

gateway/src/util/Send.ts

@@ -18,6 +18,9 @@ export async function Send(socket: WebSocket, data: Payload) {
 	}
 
 	return new Promise((res, rej) => {
+		if (socket.readyState !== 1) {
+			return rej("socket not open");
+		}
 		socket.send(buffer, (err: any) => {
 			if (err) return rej(err);
 			return res(null);

util/src/util/Event.ts

@@ -46,7 +46,9 @@ export async function listenEvent(event: string, callback: (event: EventOpts) =>
 	} else {
 		const cancel = () => {
 			events.removeListener(event, callback);
+			events.setMaxListeners(events.getMaxListeners() - 1);
 		};
+		events.setMaxListeners(events.getMaxListeners() + 1);
 		events.addListener(event, (opts) => callback({ ...opts, cancel }));
 
 		return cancel;