feat(rpc): automatically garbage collect observables + new event system + stats collection

All observables are not tracked via WeakRef and FinalizationRegistry, and frees them up on the server automatically when the client garbage collected them.

Added new event system to track and control what the rpc kernel and connections do.

Added statistics object RpcStats which collects statistics about the kernel, traffic, connections, actions, and observables.

Author: marcj

package.json

@@ -5,8 +5,8 @@
   "license": "MIT",
   "private": true,
   "scripts": {
-    "test": "node --max_old_space_size=3048 node_modules/jest/bin/jest.js --forceExit --no-cache",
-    "test:coverage": "node --max_old_space_size=3048 node_modules/jest/bin/jest.js --coverage --forceExit --no-cache",
+    "test": "node --expose-gc --max_old_space_size=3048 node_modules/jest/bin/jest.js --forceExit --no-cache",
+    "test:coverage": "node --expose-gc --max_old_space_size=3048 node_modules/jest/bin/jest.js --coverage --forceExit --no-cache",
     "build": "tsc --build tsconfig.json && tsc --build tsconfig.esm.json && lerna run build",
     "build:esm": "tsc --build tsconfig.esm.json",
     "tsc": "tsc --build",

packages/app/src/app.ts

@@ -15,10 +15,11 @@ import { AppModule, RootModuleDefinition } from './module.js';
 import { EnvConfiguration } from './configuration.js';
 import {
     DataEventToken,
+    DispatchArguments,
     EventDispatcher,
+    EventDispatcherDispatchType,
     EventListener,
     EventListenerCallback,
-    EventOfEventToken,
     EventToken,
 } from '@deepkit/event';
 import { ReceiveType, ReflectionClass, ReflectionKind } from '@deepkit/type';
@@ -292,14 +293,14 @@ export class App<T extends RootModuleDefinition> {
      *
      * order: The lower the order, the sooner the listener is called. Default is 0.
      */
-    listen<T extends EventToken<any>, DEPS extends any[]>(eventToken: T, callback: EventListenerCallback<T['event']>, order: number = 0): this {
-        const listener: EventListener<any> = { callback, order, eventToken };
+    listen<T extends EventToken<any>>(eventToken: T, callback: EventListenerCallback<T['event']>, order: number = 0): this {
+        const listener: EventListener = { callback, order, eventToken };
         this.appModule.listeners.push(listener);
         return this;
     }
 
-    public async dispatch<T extends EventToken<any>>(eventToken: T, event?: EventOfEventToken<T>, injector?: InjectorContext): Promise<void> {
-        return await this.get(EventDispatcher).dispatch(eventToken, event, injector);
+    dispatch<T extends EventToken<any>>(eventToken: T, ...args: DispatchArguments<T>): EventDispatcherDispatchType<T> {
+        return this.get(EventDispatcher).dispatch(eventToken, ...args);
     }
 
     /**

packages/app/src/module.ts

@@ -12,15 +12,7 @@ import { InjectorModule, InjectorModuleConfig, NormalizedProvider, ProviderWithS
 import { AbstractClassType, ClassType, CustomError, ExtractClassType, isClass } from '@deepkit/core';
 import { EventListener, EventToken } from '@deepkit/event';
 import { WorkflowDefinition } from '@deepkit/workflow';
-import {
-    getPartialSerializeFunction,
-    reflect,
-    ReflectionFunction,
-    ReflectionMethod,
-    serializer,
-    Type,
-    TypeClass,
-} from '@deepkit/type';
+import { getPartialSerializeFunction, reflect, ReflectionFunction, ReflectionMethod, serializer, Type, TypeClass } from '@deepkit/type';
 import { ControllerConfig } from './service-container.js';
 
 export type DefaultObject<T> = T extends undefined ? {} : T;
@@ -137,7 +129,7 @@ export interface ModuleDefinition {
      * }
      * ```
      */
-    listeners?: (EventListener<any> | ClassType)[];
+    listeners?: (EventListener | ClassType)[];
 
     /**
      * HTTP middlewares.
@@ -253,7 +245,7 @@ export function createModule<T extends CreateModuleDefinition>(options: T): AppM
     return new (createModuleClass(options))();
 }
 
-export type ListenerType = EventListener<any> | ClassType;
+export type ListenerType = EventListener | ClassType;
 
 /**
  * The AppModule is the base class for all modules.
@@ -409,7 +401,7 @@ export class AppModule<C extends InjectorModuleConfig = any> extends InjectorMod
         return this;
     }
 
-    addListener(...listener: (EventListener<any> | ClassType)[]): this {
+    addListener(...listener: (EventListener | ClassType)[]): this {
         this.assertInjectorNotBuilt();
 
         for (const l of listener) {

packages/broker/src/kernel.ts

@@ -17,6 +17,7 @@ import {
     RpcMessage,
     RpcMessageBuilder,
     RpcMessageRouteType,
+    RpcStats,
     TransportConnection,
 } from '@deepkit/rpc';
 import { Logger } from '@deepkit/logger';
@@ -52,6 +53,8 @@ import cluster from 'cluster';
 import { closeSync, openSync, renameSync, writeSync } from 'fs';
 import { snapshotState } from './snapshot.js';
 import { handleMessageDeduplication } from './utils.js';
+import { InjectorContext } from '@deepkit/injector';
+import { EventDispatcher } from '@deepkit/event';
 
 export interface Queue {
     currentId: number;
@@ -66,12 +69,15 @@ export class BrokerConnection extends RpcKernelBaseConnection {
     protected locks = new Map<number, ProcessLock>();
 
     constructor(
+        stats: RpcStats,
         logger: Logger,
         transportConnection: TransportConnection,
-        protected connections: RpcKernelConnections,
+        connections: RpcKernelConnections,
+        injector: InjectorContext,
+        eventDispatcher: EventDispatcher,
         protected state: BrokerState,
     ) {
-        super(logger, transportConnection, connections);
+        super(stats, logger, transportConnection, connections, injector, eventDispatcher);
     }
 
     public close(): void {
@@ -569,6 +575,6 @@ export class BrokerKernel extends RpcKernel {
     protected state: BrokerState = new BrokerState;
 
     createConnection(transport: TransportConnection): BrokerConnection {
-        return new BrokerConnection(this.logger, transport, this.connections, this.state);
+        return new BrokerConnection(this.stats, this.logger, transport, this.connections, this.injector, this.getEventDispatcher(), this.state);
     }
 }

packages/core/src/core.ts

@@ -604,6 +604,13 @@ export function mergeStack(error: Error, stack: string) {
     }
 }
 
+/**
+ * Makes sure the given value is an error. If it's not an error, it creates a new error with the given value as message.
+ */
+export function ensureError(error?: any, classType: ClassType = Error): Error {
+    return error instanceof Error || error instanceof AggregateError ? error : new classType(error);
+}
+
 export function collectForMicrotask<T>(callback: (args: T[]) => void): (arg: T) => void {
     let items: T[] = [];
     let taskScheduled = false;

packages/framework/src/module.ts

@@ -17,15 +17,7 @@ import { DebugDIController } from './cli/debug-di.js';
 import { ServerStartController } from './cli/server-start.js';
 import { DebugController } from './debug/debug.controller.js';
 import { registerDebugHttpController } from './debug/http-debug.controller.js';
-import {
-    http,
-    HttpLogger,
-    HttpModule,
-    HttpRegExp,
-    HttpRequest,
-    HttpResponse,
-    serveStaticListener,
-} from '@deepkit/http';
+import { http, HttpLogger, HttpModule, HttpRegExp, HttpRequest, HttpResponse, serveStaticListener } from '@deepkit/http';
 import { InjectorContext, ProviderWithScope, Token } from '@deepkit/injector';
 import { BrokerConfig, FrameworkConfig } from './module.config.js';
 import { Logger } from '@deepkit/logger';
@@ -44,15 +36,7 @@ import {
 } from '@deepkit/sql/commands';
 import { FileStopwatchStore } from './debug/stopwatch/store.js';
 import { DebugProfileFramesCommand } from './cli/debug-debug-frames.js';
-import {
-    rpcClass,
-    RpcHooks,
-    RpcKernel,
-    RpcKernelBaseConnection,
-    RpcKernelConnection,
-    RpcKernelSecurity,
-    SessionState,
-} from '@deepkit/rpc';
+import { rpcClass, RpcKernel, RpcKernelBaseConnection, RpcKernelConnection, RpcKernelSecurity, SessionState } from '@deepkit/rpc';
 import { DebugConfigController } from './cli/app-config.js';
 import { Zone } from './zone.js';
 import { DebugBrokerBus } from './debug/broker.js';
@@ -77,7 +61,6 @@ export class FrameworkModule extends createModuleClass({
         ApplicationServer,
         WebWorkerFactory,
         RpcServer,
-        RpcHooks,
         MigrationProvider,
         DebugController,
         BrokerServer,
@@ -158,7 +141,6 @@ export class FrameworkModule extends createModuleClass({
         SessionState,
         RpcKernelConnection,
         RpcKernelBaseConnection,
-        RpcHooks,
 
         BrokerDeepkitAdapter,
         BrokerCache,

packages/framework/src/rpc.ts

@@ -66,7 +66,7 @@ export class RpcServerActionWithStopwatch extends RpcServerAction {
 }
 
 export class RpcKernelConnectionWithStopwatch extends RpcKernelConnection {
-    protected actionHandler = new RpcServerActionWithStopwatch(this.cache, this, this.controllers, this.injector, this.security, this.sessionState, this.logger, this.hooks);
+    protected actionHandler = new RpcServerActionWithStopwatch(this.stats, this.cache, this, this.controllers, this.injector, this.eventDispatcher, this.security, this.sessionState, this.logger);
     stopwatch?: Stopwatch;
 
     setStopwatch(stopwatch: Stopwatch) {

packages/framework/src/worker.ts

@@ -8,13 +8,7 @@
  * You should have received a copy of the MIT License along with this program.
  */
 
-import {
-    RpcKernel,
-    RpcKernelBaseConnection,
-    RpcKernelConnection,
-    SessionState,
-    TransportConnection,
-} from '@deepkit/rpc';
+import { RpcKernel, RpcKernelBaseConnection, RpcKernelConnection, SessionState, TransportConnection } from '@deepkit/rpc';
 import http, { Server } from 'http';
 import https from 'https';
 import type { Server as WebSocketServer, ServerOptions as WebSocketServerOptions } from 'ws';
@@ -134,11 +128,15 @@ export class RpcServer implements RpcServerInterface {
                 }
             }, req);
 
-            ws.on('message', async (message: Uint8Array) => {
+            ws.on('message', (message: Uint8Array) => {
                 connection.feed(message);
             });
 
-            ws.on('close', async () => {
+            ws.on('error', (error) => {
+                connection.close(error);
+            });
+
+            ws.on('close', () => {
                 connection.close();
             });
         });

packages/http/tests/utils.ts

@@ -9,7 +9,7 @@ import { HttpRouterRegistry } from '../src/router.js';
 export function createHttpKernel(
     controllers: (ClassType | { module: AppModule<any>, controller: ClassType })[] | ((registry: HttpRouterRegistry) => void) = [],
     providers: ProviderWithScope[] = [],
-    listeners: (EventListener<any> | ClassType)[] = [],
+    listeners: (EventListener | ClassType)[] = [],
     middlewares: MiddlewareFactory[] = [],
     modules: AppModule<any>[] = []
 ) {
@@ -21,7 +21,7 @@ export function createHttpKernel(
 export function createHttpApp(
     controllers: (ClassType | { module: AppModule<any>, controller: ClassType })[] | ((registry: HttpRouterRegistry) => void) = [],
     providers: ProviderWithScope[] = [],
-    listeners: (EventListener<any> | ClassType)[] = [],
+    listeners: (EventListener | ClassType)[] = [],
     middlewares: MiddlewareFactory[] = [],
     modules: AppModule<any>[] = []
 ) {

packages/logger/src/logger.ts

@@ -335,9 +335,15 @@ export class ConsoleLogger extends Logger {
 export class MemoryLogger extends Logger {
     public memory = new MemoryLoggerTransport();
 
-    constructor() {
-        super([]);
-        this.transporter.push(this.memory);
+    constructor(
+        transporter: LoggerTransport[] = [],
+        formatter: LoggerFormatter[] = [],
+        scope: string = ''
+    ) {
+        super(transporter || [], formatter, scope);
+        if (transporter.length === 0) {
+            this.transporter.push(this.memory);
+        }
     }
 
     getOutput(): string {