feat(rpc): improve disconnect handling and cleaning up RpcMessageSubject correctly

This fixes also dangling Promises when connection closes between creating a RpcMessageSubject and fully handling it. All Promises will now always reject on disconnect/error

Author: marcj

packages/broker/src/kernel.ts

@@ -64,7 +64,6 @@ export interface Queue {
 export class BrokerConnection extends RpcKernelBaseConnection {
     protected subscribedChannels: string[] = [];
     protected locks = new Map<number, ProcessLock>();
-    protected replies = new Map<number, ((message: RpcMessage) => void)>();
 
     constructor(
         logger: Logger,
@@ -93,7 +92,7 @@ export class BrokerConnection extends RpcKernelBaseConnection {
         const promises: Promise<void>[] = [];
 
         for (const connection of this.connections.connections) {
-            if (connection === this) continue;
+            if ((connection as any) === this) continue;
             promises.push(connection.sendMessage<brokerEntityFields>(BrokerType.EntityFields, {
                 name,
                 fields,

packages/rpc/src/client/action.ts

@@ -34,9 +34,6 @@ import { EntityState, EntitySubjectStore } from './entity-state.js';
 import { assertType, deserializeType, ReflectionKind, Type, TypeObjectLiteral, typeOf } from '@deepkit/type';
 import { ProgressTracker, ProgressTrackerState } from '@deepkit/core-rxjs';
 
-interface ResponseActionObservableError extends rpcActionObservableSubscribeId, WrappedV {
-}
-
 type ControllerStateActionTypes = {
     callSchema: TypeObjectLiteral, //with args, method, and controller as property
     resultSchema: TypeObjectLiteral, //with v as property
@@ -56,7 +53,7 @@ export class RpcControllerState {
     public peerId?: string;
 
     constructor(
-        public controller: string
+        public controller: string,
     ) {
 
     }
@@ -87,7 +84,11 @@ export class RpcActionClient {
     constructor(protected client: WritableClient) {
     }
 
-    public action<T>(controller: RpcControllerState, method: string, args: any[], options: { timeout?: number, dontWaitForConnection?: true, typeReuseDisabled?: boolean } = {}) {
+    public action<T>(controller: RpcControllerState, method: string, args: any[], options: {
+        timeout?: number,
+        dontWaitForConnection?: true,
+        typeReuseDisabled?: boolean
+    } = {}) {
         const progress = ClientProgress.getNext();
 
         return asyncOperation<any>(async (resolve, reject) => {
@@ -112,12 +113,12 @@ export class RpcActionClient {
                 const subject = this.client.sendMessage(RpcTypes.Action, {
                     controller: controller.controller,
                     method: method,
-                    args
+                    args,
                 }, types.callSchema, {
                     peerId: controller.peerId,
                     dontWaitForConnection: options.dontWaitForConnection,
                     timeout: options.timeout,
-                }).onReply((reply) => {
+                }).onRejected(reject).onReply((reply) => {
                     try {
                         // console.log('client: answer', RpcTypes[reply.type], reply.composite);
 
@@ -207,7 +208,7 @@ export class RpcActionClient {
                                             unsubscribe: () => {
                                                 delete subscribers[id];
                                                 subject.send<rpcActionObservableSubscribeId>(RpcTypes.ActionObservableUnsubscribe, { id });
-                                            }
+                                            },
                                         };
                                     });
                                     (observable as any).disconnect = () => {
@@ -257,7 +258,7 @@ export class RpcActionClient {
                                         //whenever the client changes something, it's synced back to the server.
                                         //this is important to handle the stop signal.
                                         const oldChanged = observableSubject.changed;
-                                        observableSubject.changed = function (this: ProgressTracker) {
+                                        observableSubject.changed = function(this: ProgressTracker) {
                                             subject.send(RpcTypes.ActionObservableProgressNext, this.value, typeOf<ProgressTrackerState[]>());
                                             return oldChanged.apply(this);
                                         };
@@ -434,12 +435,12 @@ export class RpcActionClient {
                 const a = this.client.sendMessage<rpcActionType>(RpcTypes.ActionType, {
                     controller: controller.controller,
                     method: method,
-                    disableTypeReuse: typeReuseDisabled
+                    disableTypeReuse: typeReuseDisabled,
                 }, undefined, {
                     peerId: controller.peerId,
                     dontWaitForConnection: options.dontWaitForConnection,
                     timeout: options.timeout,
-                });
+                }).onRejected(reject);
 
                 const parsed = await a.firstThenClose<rpcResponseActionType>(RpcTypes.ResponseActionType, typeOf<rpcResponseActionType>());
 
@@ -465,8 +466,8 @@ export class RpcActionClient {
                             name: 'v',
                             parent: Object as any,
                             optional: true,
-                            type: { kind: ReflectionKind.array, type: unwrappedReturnType }
-                        }]
+                            type: { kind: ReflectionKind.array, type: unwrappedReturnType },
+                        }],
                     };
                 }
 
@@ -480,21 +481,21 @@ export class RpcActionClient {
                             { kind: ReflectionKind.propertySignature, name: 'controller', type: { kind: ReflectionKind.string } },
                             { kind: ReflectionKind.propertySignature, name: 'method', type: { kind: ReflectionKind.string } },
                             { kind: ReflectionKind.propertySignature, name: 'args', type: parameters },
-                        ]
+                        ],
                     } as TypeObjectLiteral,
                     resultSchema: {
                         kind: ReflectionKind.objectLiteral,
                         types: [
                             { kind: ReflectionKind.propertySignature, name: 'v', type: unwrappedReturnType },
-                        ]
+                        ],
                     } as TypeObjectLiteral,
                     observableNextSchema: {
                         kind: ReflectionKind.objectLiteral,
                         types: [
                             { kind: ReflectionKind.propertySignature, name: 'id', type: { kind: ReflectionKind.number } },
                             { kind: ReflectionKind.propertySignature, name: 'v', type: unwrappedReturnType },
-                        ]
-                    } as TypeObjectLiteral
+                        ],
+                    } as TypeObjectLiteral,
                 };
 
                 resolve(state.types);

packages/rpc/src/client/client-direct.ts

@@ -24,9 +24,14 @@ export class RpcDirectClientAdapter implements ClientTransportAdapter {
     }
 
     public async connect(connection: TransportClientConnection) {
+        let closed = false;
         const kernelConnection = this.rpcKernel.createConnection({
-            writeBinary: (buffer) => connection.readBinary(buffer),
+            writeBinary: (buffer) => {
+                if (closed) return;
+                connection.readBinary(buffer);
+            },
             close: () => {
+                closed = true;
                 connection.onClose('closed');
             },
         }, this.injector);
@@ -39,6 +44,7 @@ export class RpcDirectClientAdapter implements ClientTransportAdapter {
                 return 0;
             },
             close() {
+                closed = true;
                 kernelConnection.close();
             },
             writeBinary(buffer) {