Merge pull request #444 from streamich/typed-rpc-client

RPC client types

Author: streamich

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;