cleanup, automatically reconnect servers

This commit is contained in:
marlon-erler
2024-07-08 18:02:08 +02:00
parent 4daa4fd3ad
commit 1d6a1d5667
3 changed files with 36 additions and 12 deletions
+10 -9
View File
@@ -5,9 +5,11 @@ import {
forgetConnection, forgetConnection,
getWebSocketStats, getWebSocketStats,
processMessage, processMessage,
reconnectServers,
trackConnection, trackConnection,
} from "./websocket-handler"; } from "./websocket-handler";
import { import {
createFileResponse,
getVersionNumber, getVersionNumber,
writeStatNumber, writeStatNumber,
writeStatString, writeStatString,
@@ -85,11 +87,6 @@ async function main() {
})(), })(),
}); });
function createFileResponse(requestPath: string): Response {
const fullPath = Path.join(STATIC_DIR, requestPath);
return new Response(Bun.file(fullPath));
}
// OTHER SERVERS // OTHER SERVERS
connectServers(config); connectServers(config);
@@ -107,8 +104,6 @@ async function main() {
} }
// CLI // CLI
writeSuccess(`###\nstarted up ${new Date().toLocaleString()}`);
function updateCLI() { function updateCLI() {
console.clear(); console.clear();
@@ -128,8 +123,14 @@ async function main() {
console.log(); console.log();
} }
setTimeout(updateCLI, 1000); // LOOP
setInterval(updateCLI, 5000); setInterval(() => {
updateCLI();
reconnectServers(config);
}, 5000);
// FINISH
writeSuccess(`###\nstarted up ${new Date().toLocaleString()}`);
} }
main(); main();
+9 -1
View File
@@ -1,6 +1,8 @@
import { PACKAGE_FILE, STATIC_DIR } from ".";
import Colors from "colors"; import Colors from "colors";
import Fs from "fs/promises"; import Fs from "fs/promises";
import { PACKAGE_FILE } from "."; import Path from "path";
import { WebSocketMessage } from "./websocket-handler"; import { WebSocketMessage } from "./websocket-handler";
// CONFIG // CONFIG
@@ -22,6 +24,12 @@ export async function getVersionNumber(): Promise<string> {
} }
} }
// SERVER
export function createFileResponse(requestPath: string): Response {
const fullPath = Path.join(STATIC_DIR, requestPath);
return new Response(Bun.file(fullPath));
}
// OUTPUT // OUTPUT
async function writeToLogFile(message: string): Promise<void> { async function writeToLogFile(message: string): Promise<void> {
await Fs.appendFile("logs", `${message}\n`); await Fs.appendFile("logs", `${message}\n`);
+17 -2
View File
@@ -34,6 +34,7 @@ export const subscriptionMap = new SubscriptionMap();
export const clientsConnected = new Set<WS>(); export const clientsConnected = new Set<WS>();
export const serversConnectedAsClient = new Set<WS>(); export const serversConnectedAsClient = new Set<WS>();
export const serversConnected = new Set<WebSocket>(); export const serversConnected = new Set<WebSocket>();
export const serversDisconnected = new Set<string>();
export function getServerCount() { export function getServerCount() {
return serversConnected.size + serversConnectedAsClient.size; return serversConnected.size + serversConnectedAsClient.size;
@@ -45,6 +46,8 @@ export function trackConnection(ws: WS) {
export function forgetConnection(ws: WS): void { export function forgetConnection(ws: WS): void {
clientsConnected.delete(ws); clientsConnected.delete(ws);
serversConnectedAsClient.delete(ws);
const channels = subscriptionMap.getChannelList(ws); const channels = subscriptionMap.getChannelList(ws);
channels?.forEach((channel) => { channels?.forEach((channel) => {
subscriptionMap.delete(ws, channel); subscriptionMap.delete(ws, channel);
@@ -68,6 +71,12 @@ export function connectServers(config: typeof defaultConfig): void {
}); });
} }
export function reconnectServers(config: typeof defaultConfig): void {
[...serversDisconnected.values()].forEach((address) => {
connectServer(address, config);
});
}
export function connectServer( export function connectServer(
address: string, address: string,
config: typeof defaultConfig config: typeof defaultConfig
@@ -79,6 +88,8 @@ export function connectServer(
// listen // listen
ws.addEventListener("open", () => { ws.addEventListener("open", () => {
serversDisconnected.delete(address);
writeSuccess(`connected to server at "${address}".`); writeSuccess(`connected to server at "${address}".`);
subscribeChannels(ws, config); subscribeChannels(ws, config);
@@ -91,6 +102,7 @@ export function connectServer(
ws.addEventListener("close", () => { ws.addEventListener("close", () => {
serversConnected.delete(ws); serversConnected.delete(ws);
serversDisconnected.add(address);
writeError(`server at "${address}" disconnected`); writeError(`server at "${address}" disconnected`);
}); });
} catch (error) { } catch (error) {
@@ -118,6 +130,7 @@ export function getWebSocketStats(): [string, number][] {
["channels", subscriptionMap.clientsPerChannel.size], ["channels", subscriptionMap.clientsPerChannel.size],
["clients connected", clientsConnected.size], ["clients connected", clientsConnected.size],
["servers connected", getServerCount()], ["servers connected", getServerCount()],
["servers disconnected", serversDisconnected.size],
]; ];
} }
@@ -129,9 +142,11 @@ export function processMessage(
): void { ): void {
// check id // check id
const messageObject: WebSocketMessage = parseMessage(messageString); const messageObject: WebSocketMessage = parseMessage(messageString);
if (!messageObject.uuid) { // assign if no id if (!messageObject.uuid) {
// assign if no id
messageObject.uuid = Crypto.randomUUID(); messageObject.uuid = Crypto.randomUUID();
} else { // ignore if already seen } else {
// ignore if already seen
if (knownMessageIds.has(messageObject.uuid)) return; if (knownMessageIds.has(messageObject.uuid)) return;
} }
// remember message // remember message