feat(protocol): add validator cache for requestService (#333)

* fix(protocol): terminal's services is undefined * refactor(protocol): mark services and discriminator as deprecated * feat(protocol): add validator cache for requestService
No-Trade-No-Life · Dec 8, 2023 · e0be4f3 · e0be4f3
1 parent 77eb544
commit e0be4f3
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 6 deletions.
diff --git a/common/changes/@yuants/protocol/2023-12-08-16-14.json b/common/changes/@yuants/protocol/2023-12-08-16-14.json
@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@yuants/protocol",
+      "comment": "terminal's services is undefined. mark services and discriminator as deprecated. add validator cache to speed up request",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@yuants/protocol"
+}
diff --git a/libraries/protocol/etc/protocol.api.md b/libraries/protocol/etc/protocol.api.md
@@ -235,6 +235,7 @@ export interface ISubscriptionRelation {
 // @public
 export interface ITerminalInfo {
     channelIdSchemas?: JSONSchema7[];
+    // @deprecated
     discriminator?: JSONSchema7;
     env?: string;
     name?: string;
@@ -245,6 +246,7 @@ export interface ITerminalInfo {
         method: string;
         schema: JSONSchema7;
     }>;
+    // @deprecated
     services?: Array<{
         account_id?: string;
         datasource_id?: string;

diff --git a/libraries/protocol/src/model.ts b/libraries/protocol/src/model.ts
@@ -856,6 +856,8 @@ export interface ITerminalInfo {
   /**
    * List of service metadata provided by the terminal
    * 终端提供的服务元信息列表
+   *
+   * @deprecated - use serviceInfo instead
    */
   services?: Array<{ account_id?: string; datasource_id?: string; storage?: {} }>;
 
@@ -892,6 +894,8 @@ export interface ITerminalInfo {
    * If multiple terminals are candidates, the client can choose anyone to send the request.
    *
    * If the discriminator is undefined, it means that the terminal's id should be explicitly specified when sending a request.
+   *
+   * @deprecated - use serviceInfo instead
    */
   discriminator?: JSONSchema7;
 

diff --git a/libraries/protocol/src/terminal.ts b/libraries/protocol/src/terminal.ts
@@ -1,6 +1,6 @@
 import { UUID, decodePath, encodePath, formatTime } from '@yuants/data-model';
 import { batchGroupBy, rateLimitMap, switchMapWithComplete } from '@yuants/utils';
-import Ajv from 'ajv';
+import Ajv, { ValidateFunction } from 'ajv';
 import { isNode } from 'browser-or-node';
 import { JSONSchema7 } from 'json-schema';
 import {
@@ -10,6 +10,7 @@ import {
   Subscription,
   bufferCount,
   catchError,
+  combineLatestWith,
   concatMap,
   concatWith,
   defer,
@@ -31,6 +32,7 @@ import {
   shareReplay,
   takeWhile,
   tap,
+  throttleTime,
   timeout,
   timer,
   toArray,
@@ -165,6 +167,7 @@ export class Terminal {
 
     this.terminalInfo.start_timestamp_in_ms ??= Date.now();
     this.terminalInfo.status ??= 'INIT';
+    this.terminalInfo.services ??= [];
 
     this.setupReportTerminalInfo();
   }
@@ -253,11 +256,9 @@ export class Terminal {
       return this.terminalInfos$.pipe(
         first(),
         mergeMap((x) => x),
-        filter((terminalInfo) => {
-          if (!terminalInfo.serviceInfo?.[method]) return false;
-          return new Ajv({ strict: false }).validate(terminalInfo.serviceInfo[method].schema, req);
-        }),
-        map((terminalInfo) => terminalInfo.terminal_id),
+        combineLatestWith(this._mapTerminalIdAndMethodToValidator$.pipe(first())),
+        filter(([terminalInfo, map]) => map[terminalInfo.terminal_id]?.[method]?.(req)),
+        map(([terminalInfo]) => terminalInfo.terminal_id),
         toArray(),
         map((arr) => {
           const target = loadBalancer(arr, trace_id);
@@ -710,6 +711,33 @@ export class Terminal {
     shareReplay(1),
   );
 
+  private _mapTerminalIdAndMethodToValidator$: Observable<Record<string, Record<string, ValidateFunction>>> =
+    this.terminalInfos$.pipe(
+      //
+      throttleTime(30_000),
+      mergeMap((x) =>
+        from(x).pipe(
+          mergeMap((terminalInfo) =>
+            from(Object.entries(terminalInfo.serviceInfo || {})).pipe(
+              map(([method, serviceInfo]): [string, ValidateFunction] => [
+                method,
+                new Ajv().compile(serviceInfo.schema),
+              ]),
+              toArray(),
+              map((arr) => Object.fromEntries(arr)),
+              map((mapMethodToValidator): [string, Record<string, ValidateFunction>] => [
+                terminalInfo.terminal_id,
+                mapMethodToValidator,
+              ]),
+            ),
+          ),
+          toArray(),
+          map((arr) => Object.fromEntries(arr)),
+        ),
+      ),
+      shareReplay(1),
+    );
+
   /**
    * Account ID List of the same host
    */