RPC client, server, and types improvements

src/reactive-rpc/browser/createBinaryWsRpcClient.ts

@@ -4,7 +4,13 @@ import {RpcPersistentClient, WebSocketChannel} from '../common';
 import {RpcCodec} from '../common/codec/RpcCodec';
 import {BinaryRpcMessageCodec} from '../common/codec/binary';
 
-export const createBinaryWsRpcClient = (url: string) => {
+/**
+ * Constructs a JSON Reactive RPC client.
+ * @param url RPC endpoint.
+ * @param token Authentication token.
+ * @returns An RPC client.
+ */
+export const createBinaryWsRpcClient = (url: string, token: string) => {
   const writer = new Writer(1024 * 4);
   const msg = new BinaryRpcMessageCodec();
   const req = new CborJsonValueCodec(writer);
@@ -14,7 +20,7 @@ export const createBinaryWsRpcClient = (url: string) => {
     channel: {
       newChannel: () =>
         new WebSocketChannel({
-          newSocket: () => new WebSocket(url, [codec.specifier()]),
+          newSocket: () => new WebSocket(url, [codec.specifier(), token]),
         }),
     },
   });

src/reactive-rpc/browser/createJsonWsRpcClient.ts

@@ -4,7 +4,13 @@ import {RpcPersistentClient, WebSocketChannel} from '../common';
 import {RpcCodec} from '../common/codec/RpcCodec';
 import {CompactRpcMessageCodec} from '../common/codec/compact';
 
-export const createJsonWsRpcClient = (url: string) => {
+/**
+ * Constructs a JSON Reactive RPC client.
+ * @param url RPC endpoint.
+ * @param token Authentication token.
+ * @returns An RPC client.
+ */
+export const createJsonWsRpcClient = (url: string, token: string) => {
   const writer = new Writer(1024 * 4);
   const msg = new CompactRpcMessageCodec();
   const req = new JsonJsonValueCodec(writer);
@@ -14,7 +20,7 @@ export const createJsonWsRpcClient = (url: string) => {
     channel: {
       newChannel: () =>
         new WebSocketChannel({
-          newSocket: () => new WebSocket(url, [codec.specifier()]),
+          newSocket: () => new WebSocket(url, [codec.specifier(), token]),
         }),
     },
   });

src/reactive-rpc/common/codec/compact/__tests__/smoke-tests.spec.ts

@@ -84,7 +84,7 @@ for (const jsonCodec of codecList) {
     });
 
     test('Response Error typed', () => {
-      const value = RpcError.internalErrorValue();
+      const value = RpcError.internalErrorValue(null);
       const message = new ResponseErrorMessage(123, value);
       const encoded = codec.encode(jsonCodec, [message]);
       const decoded1 = jsonCodec.decoder.read(encoded);

src/reactive-rpc/common/rpc/RpcMessageBatchProcessor.ts

@@ -59,7 +59,7 @@ export class RpcMessageBatchProcessor<Ctx = unknown> {
       }
       return result;
     } catch (error) {
-      const value = RpcError.internalErrorValue();
+      const value = RpcError.internalErrorValue(error);
       return [new msg.ResponseErrorMessage(-1, value)];
     }
   }

src/reactive-rpc/common/rpc/__tests__/RpcMessageStreamProcessor.spec.ts

@@ -51,7 +51,7 @@ const setup = (
         isStreaming: false,
         call: async () => {
           // tslint:disable-next-line:no-string-throw
-          throw RpcError.internal('this promise can throw');
+          throw RpcError.internal(null, 'this promise can throw');
         },
       },
       promiseDelay: {
@@ -64,7 +64,7 @@ const setup = (
       error: {
         isStreaming: false,
         call: async () => {
-          throw RpcError.internal('this promise can throw');
+          throw RpcError.internal(null, 'this promise can throw');
         },
       },
       emitOnceSync: {
@@ -635,7 +635,7 @@ describe('pre-call checks', () => {
 
     test('fails call when pre-call checks fail', async () => {
       const onPreCall = jest.fn(async (request) => {
-        throw RpcError.internal('fail...');
+        throw RpcError.internal(null, 'fail...');
       });
       const {server, send} = setup(
         {},

src/reactive-rpc/common/rpc/__tests__/sample-api.ts

@@ -76,7 +76,7 @@ const double: IStaticRpcMethod<object, {num: number}, {num: number}> = {
 const error: IStaticRpcMethod<object, void, void> = {
   isStreaming: false,
   call: async () => {
-    throw new RpcError('this promise can throw', '', 0, '', undefined);
+    throw new RpcError('this promise can throw', '', 0, '', undefined, undefined);
   },
 };
 
@@ -96,7 +96,7 @@ const streamError: IStreamingRpcMethod<object, void, void> = {
   call$: () =>
     from(
       (async () => {
-        throw RpcError.internal('Stream always errors');
+        throw RpcError.internal(null, 'Stream always errors');
       })(),
     ),
 };

src/reactive-rpc/common/rpc/caller/RpcCaller.ts

@@ -86,9 +86,9 @@ export class RpcCaller<Ctx = unknown> {
    * @returns Response data.
    */
   public async call(name: string, request: unknown, ctx: Ctx): Promise<Value<unknown>> {
+    const method = this.getMethodStrict(name);
+    this.validate(method, request);
     try {
-      const method = this.getMethodStrict(name);
-      this.validate(method, request);
       const preCall = method.onPreCall;
       if (preCall) await preCall(ctx, request);
       const data = await method.call(request, ctx);

src/reactive-rpc/common/rpc/caller/TypeRouterCaller.ts

@@ -30,13 +30,11 @@ export class TypeRouterCaller<Router extends TypeRouter<any>, Ctx = unknown> ext
   public readonly req: {[K in keyof Routes<Router>]: MethodReq<Routes<Router>[K]>} = null as any;
   public readonly res: {[K in keyof Routes<Router>]: MethodRes<Routes<Router>[K]>} = null as any;
 
-  public get<K extends keyof Routes<Router>>(id: K): MethodDefinition<Ctx, Routes<Router>[K]> {
+  public get<K extends keyof Routes<Router>>(id: K): MethodDefinition<Ctx, Routes<Router>[K]> | undefined {
     let method = this.methods.get(id as string) as any;
     if (method) return method;
     const fn = this.router.routes[id as string];
-    // TODO: do this check without relying on constructor and importing the `FunctionType` class.
-    if (!fn || !(fn instanceof FunctionType || fn instanceof FunctionStreamingType))
-      throw RpcError.valueFromCode(RpcErrorCodes.METHOD_NOT_FOUND, `Type [alias = ${id as string}] is not a function.`);
+    if (!fn || !(fn instanceof FunctionType || fn instanceof FunctionStreamingType)) return undefined;
     const validator = fn.req.validator('object');
     const requestSchema = (fn.req as AbstractType<Schema>).getSchema();
     const isRequestVoid = requestSchema.__t === 'const' && requestSchema.value === undefined;

src/reactive-rpc/common/rpc/caller/__tests__/RpcApiCaller.spec.ts

@@ -46,7 +46,7 @@ describe('static calls', () => {
       },
     });
     const [, error] = await of(caller.call('test', {}, {}));
-    expect(error).toEqual(RpcError.internalErrorValue());
+    expect(error).toEqual(RpcError.internalErrorValue(null));
   });
 });
 
@@ -95,10 +95,10 @@ describe('streaming calls', () => {
     });
 
     const [, error1] = await of(caller.call('test', {}, {}));
-    expect(error1).toEqual(RpcError.internalErrorValue());
+    expect(error1).toEqual(RpcError.internalErrorValue(null));
 
     const [, error2] = await of(Rx.firstValueFrom(caller.call$('test', Rx.of(undefined), {})));
-    expect(error2).toEqual(RpcError.internalErrorValue());
+    expect(error2).toEqual(RpcError.internalErrorValue(null));
   });
 });
 

src/reactive-rpc/common/rpc/caller/error/RpcError.ts

@@ -36,16 +36,21 @@ export type RpcErrorValue = Value<RpcError>;
 export class RpcError extends Error implements IRpcError {
   public static from(error: unknown) {
     if (error instanceof RpcError) return error;
-    return RpcError.internal();
+    return RpcError.internal(error);
   }
 
-  public static fromCode(errno: RpcErrorCodes, message: string = '', meta: unknown = undefined): RpcError {
+  public static fromCode(
+    errno: RpcErrorCodes,
+    message: string = '',
+    meta: unknown = undefined,
+    originalError: unknown = undefined,
+  ): RpcError {
     const code = RpcErrorCodes[errno];
-    return new RpcError(message || code, code, errno, undefined, meta || undefined);
+    return new RpcError(message || code, code, errno, undefined, meta || undefined, originalError);
   }
 
-  public static internal(message: string = 'Internal Server Error'): RpcError {
-    return RpcError.fromCode(RpcErrorCodes.INTERNAL_ERROR, message);
+  public static internal(originalError: unknown, message: string = 'Internal Server Error'): RpcError {
+    return RpcError.fromCode(RpcErrorCodes.INTERNAL_ERROR, message, undefined, originalError);
   }
 
   public static badRequest(): RpcError {
@@ -60,7 +65,7 @@ export class RpcError extends Error implements IRpcError {
     return new Value(error, RpcErrorType);
   }
 
-  public static valueFrom(error: unknown, def = RpcError.internalErrorValue()): RpcErrorValue {
+  public static valueFrom(error: unknown, def = RpcError.internalErrorValue(error)): RpcErrorValue {
     if (error instanceof Value && error.data instanceof RpcError && error.type === RpcErrorType) return error;
     if (error instanceof RpcError) return RpcError.value(error);
     return def;
@@ -70,8 +75,8 @@ export class RpcError extends Error implements IRpcError {
     return RpcError.value(RpcError.fromCode(errno, message));
   }
 
-  public static internalErrorValue(): RpcErrorValue {
-    return RpcError.value(RpcError.internal());
+  public static internalErrorValue(originalError: unknown): RpcErrorValue {
+    return RpcError.value(RpcError.internal(originalError));
   }
 
   public static isRpcError(error: unknown): error is RpcError {
@@ -84,6 +89,7 @@ export class RpcError extends Error implements IRpcError {
     public readonly errno: number,
     public readonly errorId: string | undefined,
     public readonly meta: unknown | undefined,
+    public readonly originalError: unknown | undefined,
   ) {
     super(message);
     if (message === code) this.code = undefined;

src/reactive-rpc/common/rpc/caller/error/__tests__/RpcErrorType.spec.ts

@@ -6,7 +6,7 @@ import {RpcErrorType} from '../RpcErrorType';
 const codecs = new Codecs(new Writer(16));
 
 test('can encode an internal error', () => {
-  const error = RpcError.internal();
+  const error = RpcError.internal(null);
   const encoded = RpcErrorType.encode(codecs.json, error);
   // console.log(RpcErrorType.encoder(EncodingFormat.Json).toString());
   const json = JSON.parse(Buffer.from(encoded).toString());

src/reactive-rpc/common/rpc/client/types.ts

@@ -24,4 +24,26 @@ export interface RpcClient {
    * @param data Static payload data.
    */
   notify(method: string, data: undefined | unknown): void;
+
+  // start(): void;
+  // stop(): void;
+}
+
+type TypedRpcClientFn<Request, Response> = (req: Request) => Promise<Response>;
+type TypedRpcClientFn$<Request, Response> = (req: Observable<Request>) => Observable<Response>;
+type UnPromise<T> = T extends Promise<infer U> ? U : T;
+type UnObservable<T> = T extends Observable<infer U> ? U : T;
+type UnwrapResponse<T> = UnPromise<UnObservable<T>>;
+
+export interface TypedRpcClient<Routes extends Record<string, TypedRpcClientFn<any, any> | TypedRpcClientFn$<any, any>>>
+  extends RpcClient {
+  call$<K extends keyof Routes>(
+    method: K,
+    data: Parameters<Routes[K]>[0] | UnObservable<Parameters<Routes[K]>[0]>,
+  ): Observable<UnwrapResponse<ReturnType<Routes[K]>>>;
+  call<K extends keyof Routes>(
+    method: K,
+    data: Parameters<Routes[K]>[0],
+  ): Promise<UnwrapResponse<ReturnType<Routes[K]>>>;
+  notify<K extends keyof Routes>(method: K, data: UnObservable<Parameters<Routes[K]>[0]>): void;
 }

src/reactive-rpc/common/rpc/types.ts

@@ -1,3 +1,4 @@
+export * from './client/types';
 export * from './methods/types';
 export * from './caller/types';
 

src/reactive-rpc/common/testing/buildE2eClient.ts

@@ -0,0 +1,118 @@
+import {Codecs} from '../../../json-pack/codecs/Codecs';
+import {Fuzzer} from '../../../util/Fuzzer';
+import {Writer} from '../../../util/buffers/Writer';
+import {ConnectionContext} from '../../server/context';
+import {RpcCodecs} from '../codec/RpcCodecs';
+import {RpcMessageCodecs} from '../codec/RpcMessageCodecs';
+import {ReactiveRpcClientMessage, ReactiveRpcMessage, ReactiveRpcServerMessage} from '../messages';
+import {RpcMessageStreamProcessor, StreamingRpcClient, TypedRpcClient} from '../rpc';
+import type {FunctionStreamingType, FunctionType} from '../../../json-type/type/classes';
+import type {Observable} from 'rxjs';
+import type {ResolveType} from '../../../json-type';
+import type {TypeRouter} from '../../../json-type/system/TypeRouter';
+import type {TypeRouterCaller} from '../rpc/caller/TypeRouterCaller';
+
+export interface BuildE2eClientOptions {
+  /**
+   * Writer to use for encoding messages. Defaults to `new Writer(4 * 1024)`.
+   */
+  writer?: Writer;
+
+  /**
+   * Minimum and maximum size of the default buffer in kilobytes. An actual
+   * size will be picked randomly between these two values. Defaults to
+   * `[4, 4]`. Used when `writer` is not specified.
+   */
+  writerDefaultBufferKb?: [min: number, max: number];
+
+  /**
+   * Number of messages to keep in buffer before sending them to the client.
+   * The actual number of messages will be picked randomly between these two
+   * values. Defaults to `[1, 1]`.
+   */
+  serverBufferSize?: [min: number, max: number];
+
+  /**
+   * Time in milliseconds for how long to buffer messages before sending them
+   * to the client. The actual time will be picked randomly between these two
+   * values. Defaults to `[0, 0]`.
+   */
+  serverBufferTime?: [min: number, max: number];
+
+  /**
+   * Number of messages to keep in buffer before sending them to the server.
+   * The actual number of messages will be picked randomly between these two
+   * values. Defaults to `[1, 1]`.
+   */
+  clientBufferSize?: [min: number, max: number];
+
+  /**
+   * Time in milliseconds for how long to buffer messages before sending them
+   * to the server. The actual time will be picked randomly between these two
+   * values. Defaults to `[0, 0]`.
+   */
+  clientBufferTime?: [min: number, max: number];
+
+  /**
+   * IP address to use for the connection. Defaults to `0.0.0.0`.
+   */
+  ip?: string;
+
+  /**
+   * Authentication token to use for the connection. Defaults to empty string.
+   */
+  token?: string;
+}
+
+export const buildE2eClient = <Caller extends TypeRouterCaller<any>>(caller: Caller, opt: BuildE2eClientOptions) => {
+  const writer = opt.writer ?? new Writer(Fuzzer.randomInt2(opt.writerDefaultBufferKb ?? [4, 4]) * 1024);
+  const codecs = new RpcCodecs(new Codecs(writer), new RpcMessageCodecs());
+  const ctx = new ConnectionContext(
+    opt.ip ?? '0.0.0.0',
+    opt.ip ?? '',
+    null,
+    {},
+    codecs.value.cbor,
+    codecs.value.cbor,
+    codecs.messages.binary,
+  );
+  let client: StreamingRpcClient;
+  const streamProcessor = new RpcMessageStreamProcessor({
+    caller,
+    send: (messages: ReactiveRpcMessage[]) => {
+      const encoded = ctx.msgCodec.encode(ctx.resCodec, messages);
+      setTimeout(() => {
+        const decoded = ctx.msgCodec.decodeBatch(ctx.resCodec, encoded);
+        client.onMessages(decoded as ReactiveRpcServerMessage[]);
+      }, 1);
+    },
+    bufferSize: Fuzzer.randomInt2(opt.serverBufferSize ?? [1, 1]),
+    bufferTime: Fuzzer.randomInt2(opt.serverBufferTime ?? [0, 0]),
+  });
+  client = new StreamingRpcClient({
+    send: (messages: ReactiveRpcClientMessage[]) => {
+      const encoded = ctx.msgCodec.encode(ctx.reqCodec, messages);
+      setTimeout(() => {
+        const decoded = ctx.msgCodec.decodeBatch(ctx.reqCodec, encoded);
+        streamProcessor.onMessages(decoded as ReactiveRpcClientMessage[], {});
+      }, 1);
+    },
+    bufferSize: Fuzzer.randomInt2(opt.clientBufferSize ?? [1, 1]),
+    bufferTime: Fuzzer.randomInt2(opt.clientBufferTime ?? [0, 0]),
+  });
+  type Router = UnTypeRouterCaller<Caller>;
+  type Routes = UnTypeRouter<Router>;
+  type Methods = {[K in keyof Routes]: UnwrapFunction<Routes[K]>};
+  const typedClient = client as TypedRpcClient<Methods>;
+  return {
+    client: typedClient,
+  };
+};
+
+type UnTypeRouterCaller<T> = T extends TypeRouterCaller<infer R> ? R : never;
+type UnTypeRouter<T> = T extends TypeRouter<infer R> ? R : never;
+type UnwrapFunction<F> = F extends FunctionType<infer Req, infer Res>
+  ? (req: ResolveType<Req>) => Promise<ResolveType<Res>>
+  : F extends FunctionStreamingType<infer Req, infer Res>
+  ? (req$: Observable<ResolveType<Req>>) => Observable<ResolveType<Res>>
+  : never;