feat(rpc): add utility functions to create Subject/Observable synchronously from an RPC controller

Fix also race condition in DirectClient

Author: marcj

packages/rpc/index.ts

@@ -24,3 +24,4 @@ export * from './src/model.js';
 export * from './src/protocol.js';
 export * from './src/progress.js';
 export * from './src/transport.js';
+export * from './src/utils.js';

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

@@ -28,7 +28,9 @@ export class RpcDirectClientAdapter implements ClientTransportAdapter {
         const kernelConnection = this.rpcKernel.createConnection({
             writeBinary: (buffer) => {
                 if (closed) return;
-                connection.readBinary(buffer);
+                setImmediate(() => {
+                    connection.readBinary(buffer);
+                });
             },
             close: () => {
                 closed = true;

packages/rpc/src/utils.ts

@@ -0,0 +1,117 @@
+import { Observable, Subject } from 'rxjs';
+import { isFunction } from '@deepkit/core';
+
+
+/**
+ * Create a Subject with teardown function that is called when
+ * client disconnects or completes the subject.
+ *
+ * The producer is called in the next tick, no matter if the client
+ * subscribes to the subject or not. This is fundamentally different
+ * from Observables, where the producer is only called when the Observable
+ * is subscribed to.
+ *
+ * Teardown is also called when the producer errors or completes.
+ *
+ * ```typescript
+ * class Controller {
+ *   @rpc.action()
+ *   subscribeChats(channel: string): Subject<Message> {
+ *     return createSubject((subject) => {
+ *        subject.next({ text: 'hello' });
+ *        subject.next({ text: 'world' });
+ *        subject.complete();
+ *     }, () => {
+ *        // cleanup
+ *     });
+ *   }
+ * }
+ * ```
+ */
+export function createSubject<T>(
+    producer: (subject: Subject<T>) => void | Promise<void>,
+    teardown?: () => void,
+): Subject<T> {
+    const subject = new Subject<T>();
+    setImmediate(async () => {
+        try {
+            await producer(subject);
+        } catch (error) {
+            subject.error(error);
+        }
+    });
+    if (teardown) subject.subscribe().add(teardown);
+    return subject;
+}
+
+export type InstanceProducer<T> = T | Promise<T> | (() => T) | (() => Promise<T>);
+
+async function resolveProducer<T>(producer: InstanceProducer<T>): Promise<T> {
+    if (producer instanceof Promise) {
+        return producer;
+    }
+    if (isFunction(producer)) {
+        return producer();
+    }
+    return producer;
+}
+
+/**
+ * Returns a Subject that is immediately subscribed to the given producer.
+ * The producer can be a Promise, a function that returns a Promise, or a Subject.
+ *
+ * Note that the Subject will be requested from the Promise right away.
+ *
+ * ```typescript
+ * const client = new RpcClient();
+ * const controller = client.controller<MyController>('controller');
+ *
+ * // normally you would do this:
+ * const subject = await controller.subscribeChats('asd');
+ *
+ * // but with instantSubject you can do this, allowing you to get
+ * // a Subject in a synchronous way.
+ * const subject = instantSubject(controller.subscribeChats('asd'));
+ * ```
+ */
+export function instantSubject<T>(producer: InstanceProducer<Subject<T>>): Subject<T> {
+    const subject = new Subject<T>();
+    resolveProducer(producer).then((s) => {
+        s.subscribe(subject);
+        subject.subscribe().add(() => s.complete());
+    }, (error) => {
+        subject.error(error);
+    });
+    return subject;
+}
+
+/**
+ * Returns an Observable that, when subscribed to, calls the producer and subscribes to the returned Observable.
+ * The producer can be a Promise, a function that returns a Promise, or an Observable.
+ *
+ * The Observable from the RPC on the server will be automatically subscribed and unsubscribed.
+ *
+ * ```typescript
+ * const client = new RpcClient();
+ * const controller = client.controller<MyController>('controller');
+ *
+ * // normally you would do this:
+ * const observable = await controller.subscribeChats('asd');
+ *
+ * // but with instantObservable you can do this, allowing you to get
+ * // an Observable in a synchronous way.
+ * const observable = instantObservable(controller.subscribeChats('asd'));
+ *
+ * // or with a function, which is useful when you only want to do RPC calls
+ * // when the Observable is actually subscribed to.
+ * const observable = instantObservable(() => controller.subscribeChats('asd'));
+ */
+export function instantObservable<T>(producer: InstanceProducer<Observable<T>>): Observable<T> {
+    return new Observable<T>((observer) => {
+        resolveProducer(producer).then((o) => {
+            o.subscribe(observer);
+        }, (error) => {
+            observer.error(error);
+        });
+    });
+}

packages/rpc/tests/controller.spec.ts

@@ -966,15 +966,13 @@ test('connection disconnect back-controller', async () => {
         }
     }
 
-    let actionPromise: Promise<any> | undefined;
-
     class ServerController {
         constructor(private connection: RpcKernelConnection) {
         }
 
         @rpc.action()
         async start() {
-            actionPromise = this.connection.controller<ClientController>('client').action();
+            return this.connection.controller<ClientController>('client').action();
         }
     }
 
@@ -984,16 +982,14 @@ test('connection disconnect back-controller', async () => {
     const controller = client.controller<ServerController>('myController');
 
     await client.connect();
-    await controller.start();
-    await expect(actionPromise).rejects.toThrow('RpcClient has no controllers registered');
+    await expect(controller.start()).rejects.toThrow('RpcClient has no controllers registered');
 
     client.registerController(ClientController, 'client');
-    await controller.start();
-    await expect(actionPromise).resolves.toBe(true);
+    await expect(controller.start()).resolves.toBe(true);
 
-    await controller.start();
+    const promise = controller.start();
     await client.disconnect();
-    await expect(actionPromise).rejects.toThrow('Connection closed');
+    await expect(promise).rejects.toThrow('Connection closed');
 });
 
 

packages/rpc/tests/observable.spec.ts

@@ -1,11 +1,12 @@
 import { sleep } from '@deepkit/core';
 import { entity } from '@deepkit/type';
 import { expect, test } from '@jest/globals';
-import { BehaviorSubject, Observable, Subject, Subscription } from 'rxjs';
+import { BehaviorSubject, Observable, Subject, Subscription, toArray } from 'rxjs';
 import { first, take } from 'rxjs/operators';
 import { DirectClient } from '../src/client/client-direct.js';
 import { rpc } from '../src/decorators.js';
 import { RpcKernel } from '../src/server/kernel.js';
+import { createSubject, instantObservable, instantSubject } from '../src/utils';
 
 test('observable basics', async () => {
     @entity.name('model')
@@ -214,7 +215,7 @@ test('subject redirect of global subject', async () => {
             const sub = globalSubject.subscribe(subject);
             subject.subscribe().add(() => {
                 completes++;
-                sub.unsubscribe()
+                sub.unsubscribe();
             });
             return subject;
         }
@@ -279,7 +280,7 @@ test('observable unsubscribes automatically when connection closes', async () =>
                 return {
                     unsubscribe() {
                         unsubscribed = true;
-                    }
+                    },
                 };
             });
         }
@@ -465,7 +466,7 @@ test('observable complete', async () => {
                     unsubscribe() {
                         done = true;
                         active = false;
-                    }
+                    },
                 };
             });
         }
@@ -487,7 +488,7 @@ test('observable complete', async () => {
             return {
                 unsubscribe() {
                     unsubscribedCalled = true;
-                }
+                },
             };
         });
         {
@@ -537,3 +538,95 @@ test('observable complete', async () => {
     }
 });
 
+test('createSubject', async () => {
+    let teardowns = 0;
+
+    class Controller {
+        @rpc.action()
+        async subscribeChats() {
+            return createSubject<string>((subject) => {
+                subject.next('hello');
+                subject.next('world');
+            }, () => teardowns++);
+        }
+    }
+
+    const kernel = new RpcKernel();
+    kernel.registerController(Controller, 'myController');
+
+    const client = new DirectClient(kernel);
+    const controller = client.controller<Controller>('myController');
+
+    {
+        const o = await controller.subscribeChats();
+        expect(o).toBeInstanceOf(Subject);
+        const values = await o.pipe(take(2), toArray()).toPromise();
+        expect(values).toEqual(['hello', 'world']);
+        o.complete();
+        await sleep(0);
+        expect(teardowns).toBe(1);
+    }
+
+    {
+        const s = instantSubject(controller.subscribeChats());
+        const values = await s.pipe(take(2), toArray()).toPromise();
+        expect(values).toEqual(['hello', 'world']);
+        s.complete();
+        await sleep(0);
+        expect(teardowns).toBe(2);
+    }
+
+    {
+        const s = instantSubject(() => controller.subscribeChats());
+        const values = await s.pipe(take(2), toArray()).toPromise();
+        expect(values).toEqual(['hello', 'world']);
+        s.complete();
+        await sleep(0);
+        expect(teardowns).toBe(3);
+    }
+});
+
+test('instantObservable', async () => {
+    let teardowns = 0;
+
+    class Controller {
+        @rpc.action()
+        subscribeChats(channel: string) {
+            return new Observable<string>((subject) => {
+                subject.next(channel);
+                subject.next('hello');
+                subject.next('world');
+                return () => teardowns++;
+            });
+        }
+    }
+
+    const kernel = new RpcKernel();
+    kernel.registerController(Controller, 'myController');
+
+    const client = new DirectClient(kernel);
+    const controller = client.controller<Controller>('myController');
+
+    {
+        const o = await controller.subscribeChats('a');
+        expect(o).toBeInstanceOf(Observable);
+        expect(teardowns).toBe(0);
+        const values = await o.pipe(take(3), toArray()).toPromise();
+        expect(teardowns).toBe(1);
+        expect(values).toEqual(['a', 'hello', 'world']);
+    }
+
+    {
+        expect(teardowns).toBe(1);
+        const values = await instantObservable(controller.subscribeChats('b')).pipe(take(3), toArray()).toPromise();
+        expect(teardowns).toBe(2);
+        expect(values).toEqual(['b', 'hello', 'world']);
+    }
+
+    {
+        expect(teardowns).toBe(2);
+        const values = await instantObservable(() => controller.subscribeChats('b')).pipe(take(3), toArray()).toPromise();
+        expect(teardowns).toBe(3);
+        expect(values).toEqual(['b', 'hello', 'world']);
+    }
+});