feat: 实现 RPC 服务，并从 vite 迁移到 rollup

common/.eslintrc.json

@@ -10,7 +10,7 @@
   "ignorePatterns": [
     "dist",
     "coverage",
-    "vite.config.ts"
+    "rollup.config.ts"
   ],
   "rules": {
     "@typescript-eslint/explicit-function-return-type": "warn",

common/package.json

@@ -6,40 +6,43 @@
   "files": [
     "dist"
   ],
-  "main": "./dist/common.cjs",
-  "module": "./dist/common.js",
+  "main": "./dist/index.js",
+  "module": "./dist/index.js",
   "exports": {
     "types": "./dist/index.d.ts",
-    "import": "./dist/common.js",
-    "require": "./dist/common.cjs"
+    "import": "./dist/index.js",
+    "require": "./dist/index.js"
   },
   "scripts": {
-    "dev": "vite",
-    "build": "tsc && vite build",
+    "build": "rollup --config rollup.config.ts --configPlugin typescript",
     "lint": "eslint . --ext ts,tsx",
     "test": "jest",
     "coverage": "jest --coverage"
   },
   "devDependencies": {
+    "@rollup/plugin-node-resolve": "^15.2.3",
+    "@rollup/plugin-typescript": "^11.1.5",
     "@types/jest": "^29.5.5",
-    "@types/lodash": "^4.14.199",
-    "@types/node": "^20.8.3",
+    "@types/lodash": "^4.14.200",
+    "@types/node": "^20.8.7",
     "@typescript-eslint/eslint-plugin": "^6.7.4",
     "eslint": "^8.51.0",
     "eslint-config-prettier": "^9.0.0",
     "eslint-plugin-prettier": "^5.0.0",
     "jest": "^29.7.0",
-    "lodash": "^4.17.21",
     "prettier": "^3.0.3",
+    "rollup": "^4.1.4",
+    "rollup-plugin-dts": "^6.1.0",
     "ts-jest": "^29.1.1",
-    "typescript": "^5.1.3",
-    "vite": "^4.3.9",
-    "vite-plugin-dts": "^3.6.0"
+    "ts-node": "^10.9.1",
+    "tsc-alias": "^1.8.8",
+    "tsconfig-paths": "^4.2.0",
+    "typescript": "^5.1.3"
+  },
+  "dependencies": {
+    "lodash": "^4.17.21"
   },
   "_moduleAliases": {
     "@": "src"
-  },
-  "dependencies": {
-    "q": "^1.5.1"
   }
 }

common/rollup.config.ts

@@ -0,0 +1,46 @@
+import typescript from "@rollup/plugin-typescript";
+import { dts } from "rollup-plugin-dts";
+import alias from '@rollup/plugin-alias';
+import * as path from "path";
+import resolve from '@rollup/plugin-node-resolve';
+import commonjs from '@rollup/plugin-commonjs';
+
+const config = [
+  {
+    input: 'src/index.ts',
+    output: [
+      {
+        file: './dist/index.js',
+        format: 'cjs',
+        sourcemap: true,
+      }
+    ],
+    plugins: [
+      typescript({
+        tsconfig: './tsconfig.json'
+      }),
+      resolve(),
+      commonjs(),
+      alias({
+        entries: [
+          { find: '@', replacement: path.join(path.resolve(), "src") }
+        ]
+      }),
+    ]
+  },
+  {
+    input: "src/index.ts",
+    output: [{ file: "./dist/index.d.ts", format: "es" }],
+    plugins: [
+      dts(),
+      alias({
+        entries: [
+          { find: '@', replacement: path.join(path.resolve(), "src") }
+        ]
+      }),
+      resolve(),
+      commonjs()
+    ]
+  }
+]
+export default config;

common/src/Rpc/JSONFrameStream.ts

@@ -0,0 +1,66 @@
+import { Writable, WritableOptions } from "stream";
+
+export class JSONFrameStream extends Writable {
+  public handler: (obj: object) => void;
+  public frame: Frame | null;
+
+  public constructor(
+    handler: (obj: object) => void,
+    options?: WritableOptions,
+  ) {
+    super(options);
+    this.handler = handler;
+    this.frame = null;
+  }
+
+  public _write(
+    chunk: Buffer,
+    encoding: BufferEncoding,
+    callback: (error?: Error | null) => void,
+  ): void {
+    this._parse(chunk);
+    callback();
+  }
+
+  public _parse(buffer: Buffer): unknown | undefined {
+    if (!buffer.length) {
+      return;
+    }
+    if (!this.frame) {
+      this.frame = {
+        data: Buffer.alloc(4),
+        pointer: 0,
+      };
+    }
+    if (!this.frame.size) {
+      const length = Math.min(buffer.length, 4 - this.frame.pointer);
+      buffer.copy(this.frame.data, this.frame.pointer, 0, length);
+      this.frame.pointer += length;
+      if (this.frame.pointer === 4) {
+        this.frame.size = this.frame.data.readUInt32BE();
+        this.frame.data = Buffer.alloc(this.frame.size);
+        this.frame.pointer = 0;
+      }
+      return this._parse(buffer.slice(length));
+    } else {
+      const length = Math.min(
+        buffer.length,
+        this.frame.size - this.frame.pointer,
+      );
+      buffer.copy(this.frame.data, this.frame.pointer, 0, length);
+      this.frame.pointer += length;
+      if (this.frame.pointer == this.frame.size) {
+        this.handler(JSON.parse(this.frame.data.toString("utf8")));
+        this.frame = null;
+      }
+      return this._parse(buffer.slice(length));
+    }
+  }
+
+  public static makeFrame(obj: object): Buffer {
+    const data = Buffer.from(JSON.stringify(obj), "utf8");
+    const length = Buffer.alloc(4);
+    length.writeUInt32BE(data.length);
+    return Buffer.concat([length, data]);
+  }
+}

common/src/Rpc/RpcClient.ts

@@ -0,0 +1,88 @@
+import { Socket } from "net";
+import { EventEmitter } from "events";
+import { JSONFrameStream } from "@/Rpc/JSONFrameStream";
+
+export class RpcClient {
+  public socket: Socket;
+  public requestId: number;
+  public defers: Map<
+    string,
+    {
+      defer: Promise<unknown>;
+      reject: (reason?: Error | undefined) => void;
+      resolve: (value?: object) => void;
+    }
+  >;
+  public pubSub: EventEmitter;
+
+  public constructor(socket: Socket) {
+    this.socket = socket;
+    this.socket.pipe(
+      new JSONFrameStream(
+        (this._processFrame as (obj: object) => void).bind(this),
+      ),
+    );
+    this.requestId = 0;
+    this.defers = new Map();
+    this.pubSub = new EventEmitter();
+  }
+
+  public _processFrame(obj: RpcClientFrameObj): void {
+    if (obj.pubSub) {
+      this.pubSub.emit(obj.pubSub.channel, obj.pubSub.channel, obj.pubSub.data);
+      this.pubSub.emit("*", obj.pubSub.channel, obj.pubSub.data);
+      return;
+    }
+    if (!this.defers.has(String(obj.id))) {
+      console.error("invalid request id", obj.id);
+      return;
+    }
+    if (obj.error) {
+      this.defers.get(String(obj.id))!.reject(obj.error);
+    } else {
+      this.defers.get(String(obj.id))!.resolve(obj.result);
+    }
+    this.defers.delete(String(obj.id));
+  }
+
+  public request(
+    method: string,
+    ...args: unknown[]
+  ): {
+    defer: Promise<unknown>;
+    reject: (reason?: Error) => void;
+    resolve: (value?: object) => void;
+  } {
+    this.requestId++;
+    const request = {
+      id: this.requestId,
+      method,
+      args,
+    };
+    let resolve = ((): void => {}) as (value?: object) => void,
+      reject = ((): void => {}) as (reason?: Error) => void;
+    this.socket.write(JSONFrameStream.makeFrame(request));
+    const defer = new Promise((res, rej) => {
+      [resolve, reject] = [res, rej];
+    });
+    this.defers.set(String(this.requestId), { defer, reject, resolve });
+    return { defer, reject, resolve };
+  }
+
+  public subscribe(
+    channelToSubscribe: string,
+    callback: (...args: unknown[]) => void,
+  ): void {
+    const request = {
+      method: "subscribe",
+      channel: channelToSubscribe,
+    };
+    this.socket.write(JSONFrameStream.makeFrame(request));
+    this.pubSub.addListener(
+      channelToSubscribe,
+      (channel: string, ...args: unknown[]) => {
+        callback.apply({ channel }, args);
+      },
+    );
+  }
+}

common/src/Rpc/RpcServer.ts

@@ -0,0 +1,57 @@
+import { Socket } from "net";
+import { JSONFrameStream } from "@/Rpc/JSONFrameStream";
+
+export class RpcServer {
+  public socket: Socket;
+  public methods: RpcMethods;
+  public channelUnsubscribe: Map<string, () => void>;
+  public constructor(socket: Socket, methods: RpcMethods) {
+    this.socket = socket;
+    this.socket.pipe(
+      new JSONFrameStream(
+        (this._processFrame as (obj: object) => void).bind(this),
+      ),
+    );
+    this.methods = methods;
+    this.channelUnsubscribe = new Map();
+    this.socket.on("close", () => {
+      this.channelUnsubscribe.forEach((unsubscribe) => unsubscribe());
+      this.channelUnsubscribe.clear();
+    });
+  }
+
+  public _processFrame(obj: RpcServerFrameObj): void {
+    const args = obj.args || [];
+    if (obj.method === "subscribe") {
+      if (this.channelUnsubscribe.has("*")) {
+        return;
+      }
+      if (obj.channel === "*") {
+        this.channelUnsubscribe.forEach((unsubscribe) => unsubscribe());
+        this.channelUnsubscribe.clear();
+      }
+      if (!this.channelUnsubscribe.has(obj.channel)) {
+        const unsubscribe = this.methods.subscribe(obj.channel, (pubSub) => {
+          this.socket.write(JSONFrameStream.makeFrame({ pubSub }));
+        });
+        this.channelUnsubscribe.set(obj.channel, unsubscribe);
+      }
+      this.socket.write("");
+      return;
+    }
+    this.methods[obj.method].apply(
+      null,
+      args.concat([
+        (error: Error, result: object): void => {
+          const response: RpcResponse = { id: obj.id };
+          if (error) {
+            response.error = error;
+          } else {
+            response.result = result;
+          }
+          this.socket.write(JSONFrameStream.makeFrame(response));
+        },
+      ]),
+    );
+  }
+}

common/src/index.ts

@@ -1,4 +1,6 @@
 import { ConfigManager } from "@/ConfigManager";
+export { RpcServer } from "@/Rpc/RpcServer";
+export { RpcClient } from "@/Rpc/RpcClient";
 
 export const common = {
   configManager: new ConfigManager(),

common/src/types.d.ts

@@ -0,0 +1,30 @@
+type Frame = { data: Buffer; pointer: number; size?: number };
+
+type RpcServerFrameObj = {
+  method: string;
+  channel: string;
+  id: string;
+  args: unknown[];
+};
+
+type RpcClientFrameObj = {
+  id: string;
+  error?: Error;
+  result?: object;
+  pubSub: ChildProcess;
+};
+
+type RpcMethods = {
+  subscribe(
+    channel: string,
+    listener: (data: { channel: string; data: unknown }) => void,
+  ): () => void;
+  publish: (channel: string, data: unknown, cb?: CallBack) => void;
+  [name: string]: (cb?: CallBack, ...args) => void;
+};
+
+type RpcResponse = {
+  id: string;
+  error?: Error;
+  result?: object;
+};

common/tests/Rpc/JSONFrameStream.test.ts

@@ -0,0 +1,7 @@
+import { JSONFrameStream } from "@/Rpc/JSONFrameStream";
+
+test("JSONFrameStream", () => {
+  const buffer = JSONFrameStream.makeFrame({ Loki: "god" });
+  console.log(buffer);
+  expect(buffer.length).toBe(18);
+});