feat(reactive-rpc): 🎸 implement server builder

Author: streamich

src/reactive-rpc/__demos__/ws.ts

@@ -1,29 +1,10 @@
 // npx ts-node src/reactive-rpc/__demos__/ws.ts
 
-import {Http1Server} from '../server/http1/Http1Server';
+import {createCaller} from '../common/rpc/__tests__/sample-api';
+import {RpcServer} from '../server/http1/RpcServer';
 
-const server = Http1Server.start();
-
-server.ws({
-  path: '/ws',
-  onConnect: (connection) => {
-    console.log('CONNECTED');
-    connection.onmessage = (data, isUtf8) => {
-      console.log('MESSAGE', data, isUtf8);
-    };
-  },
-});
-
-server.route({
-  path: '/hello',
-  handler: ({res}) => {
-    res.statusCode = 200;
-    res.end('Hello World\n');
-  },
+RpcServer.startWithDefaults({
+  port: 3000,
+  caller: createCaller(),
+  logger: console,
 });
-
-server.enableHttpPing();
-
-server.start();
-
-console.log(server + '');

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

@@ -34,7 +34,7 @@ export enum RpcErrorCodes {
 export type RpcErrorValue = RpcValue<RpcError>;
 
 export class RpcError extends Error implements IRpcError {
-  public static from(error: unknown) {
+  public static from(error: unknown): RpcError {
     if (error instanceof RpcError) return error;
     return RpcError.internal(error);
   }

src/reactive-rpc/server/http1/Http1Server.ts

@@ -135,7 +135,7 @@ export class Http1Server implements Printable {
       const codecs = this.codecs;
       const ip = this.findIp(req);
       const token = this.findToken(req);
-      const ctx = new Http1ConnectionContext(res, path, query, ip, token, match.params, new NullObject(), codecs.value.json, codecs.value.json, codecs.messages.jsonRpc2);
+      const ctx = new Http1ConnectionContext(req, res, path, query, ip, token, match.params, new NullObject(), codecs.value.json, codecs.value.json, codecs.messages.jsonRpc2);
       const handler = match.data.handler;
       handler(ctx);
     } catch (error) {

src/reactive-rpc/server/http1/RpcServer.ts

@@ -0,0 +1,123 @@
+import * as http from 'http';
+import {Printable} from '../../../util/print/types';
+import {printTree} from '../../../util/print/printTree';
+import {Http1Server} from './Http1Server';
+import {RpcError} from '../../common/rpc/caller';
+import {IncomingBatchMessage, RpcMessageBatchProcessor} from '../../common';
+import {ConnectionContext} from './context';
+import type {RpcCaller} from '../../common/rpc/caller/RpcCaller';
+import type {ServerLogger} from './types';
+
+const DEFAULT_MAX_PAYLOAD = 4 * 1024 * 1024;
+
+export interface RpcServerOpts {
+  http1: Http1Server;
+  caller: RpcCaller<any>;
+  logger?: ServerLogger;
+}
+
+export interface RpcServerStartOpts extends Omit<RpcServerOpts, 'http1'> {
+  port?: number;
+  server?: http.Server;
+}
+
+export class RpcServer implements Printable {
+  public static readonly create = (opts: RpcServerOpts) => {
+    const server = new RpcServer(opts);
+    opts.http1.enableHttpPing();
+    return server;
+  };
+
+  public static readonly startWithDefaults = (opts: RpcServerStartOpts) => {
+    const port = opts.port ?? 8080;
+    const logger = opts.logger ?? console;
+    const server = http.createServer();
+    const http1Server = new Http1Server({
+      server,
+    });
+    const rpcServer = RpcServer.create({
+      caller: opts.caller,
+      http1: http1Server,
+      logger,
+    });
+    rpcServer.enableDefaults();
+    http1Server.start();
+    server.listen(port, () => {
+      let host = server.address() || 'localhost';
+      if (typeof host === 'object') host = (host as any).address;
+      logger.log({msg: 'SERVER_STARTED', host, port});
+    });
+  };
+
+  public readonly http1: Http1Server;
+  protected readonly batchProcessor: RpcMessageBatchProcessor<ConnectionContext>;
+
+  constructor (protected readonly opts: RpcServerOpts) {
+    const http1 = this.http1 = opts.http1;
+    const onInternalError = http1.oninternalerror;
+    http1.oninternalerror = (error, res, req) => {
+      if (error instanceof RpcError) {
+        res.statusCode = 400;
+        const data = JSON.stringify(error.toJson());
+        res.end(data);
+        return;
+      }
+      onInternalError(error, res, req);
+    };
+    this.batchProcessor = new RpcMessageBatchProcessor<ConnectionContext>({caller: opts.caller});
+  }
+
+  public enableHttpPing(): void {
+    this.http1.enableHttpPing();
+  }
+
+  public enableHttpRpc(path: string = '/rpc'): void {
+    const batchProcessor = this.batchProcessor;
+    const logger = this.opts.logger ?? console;
+    this.http1.route({
+      method: 'POST',
+      path,
+      handler: async (ctx) => {
+        const res = ctx.res;
+        const body = await ctx.body(DEFAULT_MAX_PAYLOAD);
+        if (!res.socket) return;
+        try {
+          const messageCodec = ctx.msgCodec;
+          const incomingMessages = messageCodec.decodeBatch(ctx.reqCodec, body);
+          try {
+            const outgoingMessages = await batchProcessor.onBatch(incomingMessages as IncomingBatchMessage[], ctx);
+            if (!res.socket) return;
+            const resCodec = ctx.resCodec;
+            messageCodec.encodeBatch(resCodec, outgoingMessages);
+            const buf = resCodec.encoder.writer.flush();
+            if (!res.socket) return;
+            res.end(buf);
+          } catch (error) {
+            logger.error('HTTP_RPC_PROCESSING', error, {messages: incomingMessages});
+            throw RpcError.from(error);
+          }
+        } catch (error) {
+          if (typeof error === 'object' && error)
+            if ((error as any).message === 'Invalid JSON') throw RpcError.badRequest();
+          throw RpcError.from(error);
+        }
+      },
+    });
+  }
+
+  public enableDefaults(): void {
+    // this.enableCors();
+    this.enableHttpPing();
+    this.enableHttpRpc();
+    // this.enableWsRpc();
+    // this.startRouting();
+  }
+
+  // ---------------------------------------------------------------- Printable
+
+  public toString(tab: string = ''): string {
+    return `${this.constructor.name}` + printTree(tab, [
+      (tab) => `HTTP/1.1 ${this.http1.toString(tab)}`,
+    ]);
+  }
+}

src/reactive-rpc/server/http1/context.ts

@@ -1,3 +1,5 @@
+import {getBody} from './util';
+import {listToUint8} from '../../../util/buffers/concat';
 import type * as http from 'http';
 import type {JsonValueCodec} from '../../../json-pack/codecs/types';
 import type {RpcMessageCodec} from '../../common/codec/types';
@@ -19,6 +21,7 @@ export interface ConnectionContext<Meta = Record<string, unknown>> {
 
 export class Http1ConnectionContext<Meta = Record<string, unknown>> implements ConnectionContext<Meta> {
   constructor(
+    public readonly req: http.IncomingMessage,
     public readonly res: http.ServerResponse,
     public path: string,
     public query: string,
@@ -88,4 +91,10 @@ export class Http1ConnectionContext<Meta = Record<string, unknown>> implements C
       }
     }
   }
+
+  public async body(maxPayload: number): Promise<Uint8Array> {
+    const list = await getBody(this.req, maxPayload);
+    const bodyUint8 = listToUint8(list);
+    return bodyUint8;
+  }
 }

src/reactive-rpc/server/http1/types.ts

@@ -0,0 +1,4 @@
+export interface ServerLogger {
+  log(msg: unknown): void;
+  error(kind: string, error?: Error | unknown | null, meta?: unknown): void;
+}