core: fix memory leak

This commit is contained in:
undefined committed 2024-02-20 22:22:59 +08:00
1 parent 846a394e54
commit 0a944b22db
2 files changed
+13 -7

No files matched your search

+6 -3
View File
@@ -14,7 +14,7 @@ export class WebSocketLayer {
clients = new Set<WebSocket>(); clients = new Set<WebSocket>();
regexp: RegExp; regexp: RegExp;
constructor(private router: Router, path: Parameters<typeof pathToRegexp>[0], public callback?: WebSocketCallback) { constructor(path: Parameters<typeof pathToRegexp>[0], public callback?: WebSocketCallback) {
this.regexp = pathToRegexp(path); this.regexp = pathToRegexp(path);
} }
@@ -49,9 +49,12 @@ export class Router extends KoaRouter {
} }
ws(path: Parameters<typeof pathToRegexp>[0], callback?: WebSocketCallback) { ws(path: Parameters<typeof pathToRegexp>[0], callback?: WebSocketCallback) {
const layer = new WebSocketLayer(this, path, callback); const layer = new WebSocketLayer(path, callback);
this.wsStack.push(layer); this.wsStack.push(layer);
this.disposeLastOp = () => remove(this.wsStack, layer); this.disposeLastOp = () => {
layer.close();
remove(this.wsStack, layer);
};
return layer; return layer;
} }
} }
+7 -4
View File
@@ -414,7 +414,7 @@ export function Connection(
...permPrivChecker: Array<number | bigint | Function> ...permPrivChecker: Array<number | bigint | Function>
) { ) {
const checker = Checker(permPrivChecker); const checker = Checker(permPrivChecker);
router.ws(prefix, async (conn, _, ctx) => { const layer = router.ws(prefix, async (conn, _, ctx) => {
const { const {
args, request, response, user, domain, UiContext, args, request, response, user, domain, UiContext,
} = ctx.HydroContext; } = ctx.HydroContext;
@@ -437,6 +437,7 @@ export function Connection(
if (closed) return; if (closed) return;
closed = true; closed = true;
bus.emit('connection/close', h); bus.emit('connection/close', h);
layer.clients.delete(conn);
if (interval) clearInterval(interval); if (interval) clearInterval(interval);
for (const d of disposables) d(); for (const d of disposables) d();
h.cleanup?.(args); h.cleanup?.(args);
@@ -446,7 +447,7 @@ export function Connection(
clean(); clean();
conn.terminate(); conn.terminate();
} }
if (Date.now() - lastHeartbeat > 30000) conn.ping(); if (Date.now() - lastHeartbeat > 30000) conn.send('ping');
}, 40000); }, 40000);
conn.on('pong', () => { conn.on('pong', () => {
lastHeartbeat = Date.now(); lastHeartbeat = Date.now();
@@ -460,9 +461,11 @@ export function Connection(
} }
h.message?.(JSON.parse(e.data.toString())); h.message?.(JSON.parse(e.data.toString()));
}; };
conn.onclose = clean;
conn.resume();
await bus.parallel('connection/active', h); await bus.parallel('connection/active', h);
if (conn.readyState === conn.OPEN) {
conn.on('close', clean);
conn.resume();
} else clean();
} catch (e) { } catch (e) {
await h.onerror(e); await h.onerror(e);
} }