feat(rpc): auto-GC observables, event system, statistics

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/event/README.md

@@ -1,3 +1,3 @@
 # @deepkit/event
 
-Deepkit async event dispatcher.
+Deepkit async/sync event dispatcher.

packages/event/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@deepkit/event",
   "version": "1.0.2",
-  "description": "Deepkit async event dispatcher",
+  "description": "Deepkit event dispatcher",
   "type": "commonjs",
   "main": "./dist/cjs/index.js",
   "module": "./dist/esm/index.js",