add DataPack encode and decode rpc message

elpheria · Nov 17, 2023 · c7a07fc · c7a07fc
1 parent c9e6c63
commit c7a07fc
Show file tree

Hide file tree

Showing 17 changed files with 221 additions and 60 deletions.
diff --git a/API.md b/API.md
@@ -74,6 +74,7 @@ Parameters:
   * `max_reconnects` {Number}: Maximum number of times the client should try to reconnect. Defaults to `5`. `0` means unlimited.
   * Any other option allowed in <a href="https://github.com/websockets/ws/blob/master/doc/ws.md#new-websocketaddress-protocols-options" target="_blank">Node WebSocket</a>
 * `generate_request_id` {Function} Custom function to generate request id instead of simple increment by default. Passes `method` and `params` to parameters.
+* `dataPack` {DataPack} data pack contains encoder and decoder.
 
 ### ws.connect()
 
@@ -180,6 +181,8 @@ Parameters:
 * `options` {Object}: Server options that are also forwarded to `ws`.
   * `port` {Number}: Port number on which the server will listen for incoming requests.
   * `host` {String}: Address on which the server will listen for incoming requests.
+* `dataPack` {DataPack} data pack contains encoder and decoder.
+
 
 Once the Server class is instantiated, you can use a `ws` library's instance via server.wss object.
 

diff --git a/build-ts/lib/client.d.ts b/build-ts/lib/client.d.ts
@@ -5,6 +5,7 @@
  */
 import { EventEmitter } from "eventemitter3";
 import { NodeWebSocketType, ICommonWebSocketFactory } from "./client/client.types";
+import { DataPack } from "./utils";
 interface IQueueElement {
     promise: [
         Parameters<ConstructorParameters<typeof Promise>[0]>[0],
@@ -35,21 +36,23 @@ export default class CommonClient extends EventEmitter {
     private generate_request_id;
     private socket;
     private webSocketFactory;
+    private dataPack;
     /**
      * Instantiate a Client class.
      * @constructor
      * @param {webSocketFactory} webSocketFactory - factory method for WebSocket
      * @param {String} address - url to a websocket server
      * @param {Object} options - ws options object with reconnect parameters
      * @param {Function} generate_request_id - custom generation request Id
+     * @param {DataPack} dataPack - data pack contains encoder and decoder
      * @return {CommonClient}
      */
     constructor(webSocketFactory: ICommonWebSocketFactory, address?: string, { autoconnect, reconnect, reconnect_interval, max_reconnects, ...rest_options }?: {
         autoconnect?: boolean;
         reconnect?: boolean;
         reconnect_interval?: number;
         max_reconnects?: number;
-    }, generate_request_id?: (method: string, params: object | Array<any>) => number);
+    }, generate_request_id?: (method: string, params: object | Array<any>) => number, dataPack?: DataPack<object, string>);
     /**
      * Connects to a defined server if not connected already.
      * @method

diff --git a/build-ts/lib/client.js b/build-ts/lib/client.js
@@ -17,6 +17,7 @@ var __rest = (this && this.__rest) || function (s, e) {
 };
 // @ts-ignore
 import { EventEmitter } from "eventemitter3";
+import { DefaultDataPack } from "./utils";
 export default class CommonClient extends EventEmitter {
     /**
      * Instantiate a Client class.
@@ -25,9 +26,10 @@ export default class CommonClient extends EventEmitter {
      * @param {String} address - url to a websocket server
      * @param {Object} options - ws options object with reconnect parameters
      * @param {Function} generate_request_id - custom generation request Id
+     * @param {DataPack} dataPack - data pack contains encoder and decoder
      * @return {CommonClient}
      */
-    constructor(webSocketFactory, address = "ws://localhost:8080", _a = {}, generate_request_id) {
+    constructor(webSocketFactory, address = "ws://localhost:8080", _a = {}, generate_request_id, dataPack) {
         var { autoconnect = true, reconnect = true, reconnect_interval = 1000, max_reconnects = 5 } = _a, rest_options = __rest(_a, ["autoconnect", "reconnect", "reconnect_interval", "max_reconnects"]);
         super();
         this.webSocketFactory = webSocketFactory;
@@ -43,6 +45,10 @@ export default class CommonClient extends EventEmitter {
         this.rest_options = rest_options;
         this.current_reconnects = 0;
         this.generate_request_id = generate_request_id || (() => ++this.rpc_id);
+        if (!dataPack)
+            this.dataPack = new DefaultDataPack();
+        else
+            this.dataPack = dataPack;
         if (this.autoconnect)
             this._connect(this.address, Object.assign({ autoconnect: this.autoconnect, reconnect: this.reconnect, reconnect_interval: this.reconnect_interval, max_reconnects: this.max_reconnects }, this.rest_options));
     }
@@ -80,7 +86,7 @@ export default class CommonClient extends EventEmitter {
                 params: params || null,
                 id: rpc_id
             };
-            this.socket.send(JSON.stringify(message), ws_opts, (error) => {
+            this.socket.send(this.dataPack.encode(message), ws_opts, (error) => {
                 if (error)
                     return reject(error);
                 this.queue[rpc_id] = { promise: [resolve, reject] };
@@ -129,7 +135,7 @@ export default class CommonClient extends EventEmitter {
                 method: method,
                 params: params || null
             };
-            this.socket.send(JSON.stringify(message), (error) => {
+            this.socket.send(this.dataPack.encode(message), (error) => {
                 if (error)
                     return reject(error);
                 resolve();
@@ -196,7 +202,7 @@ export default class CommonClient extends EventEmitter {
             if (message instanceof ArrayBuffer)
                 message = Buffer.from(message).toString();
             try {
-                message = JSON.parse(message);
+                message = this.dataPack.decode(message);
             }
             catch (error) {
                 return;

diff --git a/build-ts/lib/server.d.ts b/build-ts/lib/server.d.ts
@@ -4,6 +4,7 @@
  */
 import { EventEmitter } from "eventemitter3";
 import NodeWebSocket, { Server as WebSocketServer } from "ws";
+import * as utils from "./utils";
 interface INamespaceEvent {
     [x: string]: {
         sockets: Array<string>;
@@ -33,14 +34,16 @@ interface IClientWebSocket extends NodeWebSocket {
 }
 export default class Server extends EventEmitter {
     private namespaces;
+    private dataPack;
     wss: InstanceType<typeof WebSocketServer>;
     /**
      * Instantiate a Server class.
      * @constructor
      * @param {Object} options - ws constructor's parameters with rpc
+     * @param {DataPack} dataPack - data pack contains encoder and decoder
      * @return {Server} - returns a new Server instance
      */
-    constructor(options: NodeWebSocket.ServerOptions);
+    constructor(options: NodeWebSocket.ServerOptions, dataPack?: utils.DataPack<object, string>);
     /**
      * Registers an RPC method.
      * @method

diff --git a/build-ts/lib/server.js b/build-ts/lib/server.js
@@ -14,9 +14,10 @@ export default class Server extends EventEmitter {
      * Instantiate a Server class.
      * @constructor
      * @param {Object} options - ws constructor's parameters with rpc
+     * @param {DataPack} dataPack - data pack contains encoder and decoder
      * @return {Server} - returns a new Server instance
      */
-    constructor(options) {
+    constructor(options, dataPack) {
         super();
         /**
          * Stores all connected sockets with a universally unique identifier
@@ -30,6 +31,10 @@ export default class Server extends EventEmitter {
          * @param {Object} namespaces.events
          */
         this.namespaces = {};
+        if (!dataPack)
+            this.dataPack = new utils.DefaultDataPack();
+        else
+            this.dataPack = dataPack;
         this.wss = new WebSocketServer(options);
         this.wss.on("listening", () => this.emit("listening"));
         this.wss.on("connection", (socket, request) => {
@@ -180,7 +185,7 @@ export default class Server extends EventEmitter {
                 const socket = this.namespaces[ns].clients.get(socket_id);
                 if (!socket)
                     continue;
-                socket.send(JSON.stringify({
+                socket.send(this.dataPack.encode({
                     notification: name,
                     params: params || null
                 }));
@@ -236,7 +241,7 @@ export default class Server extends EventEmitter {
             emit(event, ...params) {
                 const socket_ids = [...self.namespaces[name].clients.keys()];
                 for (let i = 0, id; id = socket_ids[i]; ++i) {
-                    self.namespaces[name].clients.get(id).send(JSON.stringify({
+                    self.namespaces[name].clients.get(id).send(self.dataPack.encode({
                         notification: event,
                         params: params || []
                     }));
@@ -335,18 +340,18 @@ export default class Server extends EventEmitter {
                 return; // TODO: should have debug logs here
             let parsedData;
             try {
-                parsedData = JSON.parse(data);
+                parsedData = this.dataPack.decode(data);
             }
             catch (error) {
-                return socket.send(JSON.stringify({
+                return socket.send(this.dataPack.encode({
                     jsonrpc: "2.0",
                     error: utils.createError(-32700, error.toString()),
                     id: null
                 }), msg_options);
             }
             if (Array.isArray(parsedData)) {
                 if (!parsedData.length)
-                    return socket.send(JSON.stringify({
+                    return socket.send(this.dataPack.encode({
                         jsonrpc: "2.0",
                         error: utils.createError(-32600, "Invalid array"),
                         id: null
@@ -360,12 +365,12 @@ export default class Server extends EventEmitter {
                 }
                 if (!responses.length)
                     return;
-                return socket.send(JSON.stringify(responses), msg_options);
+                return socket.send(this.dataPack.encode(responses), msg_options);
             }
             const response = await this._runMethod(parsedData, socket._id, ns);
             if (!response)
                 return;
-            return socket.send(JSON.stringify(response), msg_options);
+            return socket.send(this.dataPack.encode(response), msg_options);
         });
     }
     /**

diff --git a/build-ts/lib/utils.d.ts b/build-ts/lib/utils.d.ts
@@ -3,6 +3,14 @@ interface IRPCError {
     message: string;
     data?: string;
 }
+export interface DataPack<T, R extends string | ArrayBufferLike | Blob | ArrayBufferView> {
+    encode(value: T): R;
+    decode(value: R): T;
+}
+export declare class DefaultDataPack implements DataPack<Object, string> {
+    encode(value: Object): string;
+    decode(value: string): Object;
+}
 /**
  * Creates a JSON-RPC 2.0-compliant error.
  * @param {Number} code - error code

diff --git a/build-ts/lib/utils.js b/build-ts/lib/utils.js
@@ -10,6 +10,14 @@ const errors = new Map([
     [-32606, "Event forbidden"],
     [-32700, "Parse error"]
 ]);
+export class DefaultDataPack {
+    encode(value) {
+        return JSON.stringify(value);
+    }
+    decode(value) {
+        return JSON.parse(value);
+    }
+}
 /**
  * Creates a JSON-RPC 2.0-compliant error.
  * @param {Number} code - error code