feat(rpc/client): batches requests

rafamel · Nov 1, 2019 · e5aa664 · e5aa664
1 parent 95ccfff
commit e5aa664
Show file tree

Hide file tree

Showing 7 changed files with 37 additions and 8 deletions.
diff --git a/packages/rpc/package-lock.json b/packages/rpc/package-lock.json
diff --git a/packages/rpc/package.json b/packages/rpc/package.json
@@ -79,6 +79,7 @@
   "dependencies": {
     "@karmic/core": "^0.1.0",
     "ajv": "^6.10.2",
+    "dataloader": "^1.4.0",
     "promist": "^2.0.0"
   },
   "peerDependencies": {

diff --git a/packages/rpc/src/client/ClientManager/ClientManager.ts b/packages/rpc/src/client/ClientManager/ClientManager.ts
@@ -25,11 +25,15 @@ export class ClientManager {
   public constructor(
     connection: RPCClientConnection,
     parser: DataParser,
-    subscriptionConnectRetry: boolean
+    subscriptionConnectRetry: boolean,
+    batchRequests: boolean
   ) {
     this.store = new ClientStore();
-    this.connection = new ConnectionManager(connection, parser, (data) =>
-      this.response(data)
+    this.connection = new ConnectionManager(
+      connection,
+      parser,
+      (data) => this.response(data),
+      batchRequests
     );
     this.subscriptionConnectRetry = subscriptionConnectRetry;
     this.subscription = this.connection.status$.subscribe((status) => {

diff --git a/packages/rpc/src/client/ClientManager/ConnectionManager.ts b/packages/rpc/src/client/ClientManager/ConnectionManager.ts
@@ -2,6 +2,7 @@ import { RPCClientConnection, RPCClientStatus } from '../types';
 import { BehaviorSubject, Subject, Observable, Subscription } from 'rxjs';
 import { DataInput, DataParser, RPCRequest, RPCNotification } from '~/types';
 import { until } from 'promist';
+import DataLoader from 'dataloader';
 
 interface ConnectionManagerSubjects {
   status: BehaviorSubject<RPCClientStatus>;
@@ -11,18 +12,29 @@ interface ConnectionManagerSubjects {
 export class ConnectionManager {
   public errors: Error[];
   private connection: RPCClientConnection;
-  private parser: DataParser;
+  private loader: DataLoader<RPCRequest | RPCNotification, null | Error>;
   private subjects: ConnectionManagerSubjects;
   private subscription: Subscription;
   public constructor(
     connection: RPCClientConnection,
     parser: DataParser,
-    onData: (data: object) => void
+    onData: (data: object) => void,
+    batchRequests: boolean
   ) {
     this.errors = [];
 
-    this.parser = parser;
     this.connection = connection;
+    this.loader = new DataLoader<RPCRequest | RPCNotification, null | Error>(
+      async (arr) => {
+        try {
+          await this.connection.actions.send(await parser.serialize(arr));
+          return Array(arr.length).fill(null);
+        } catch (err) {
+          return Array(arr.length).fill(err);
+        }
+      },
+      { batch: batchRequests, cache: false }
+    );
     this.subjects = {
       status: new BehaviorSubject<RPCClientStatus>(connection.status),
       errors: new Subject<Error>()
@@ -82,7 +94,8 @@ export class ConnectionManager {
   }
   public async send(data: RPCRequest | RPCNotification): Promise<void> {
     if (this.status === 'complete') return;
-    await this.connection.actions.send(await this.parser.serialize(data));
+
+    await this.loader.load(data);
   }
   public report(error: Error): void {
     if (this.status === 'complete') return;

diff --git a/packages/rpc/src/client/RPCClient.ts b/packages/rpc/src/client/RPCClient.ts
@@ -22,7 +22,8 @@ export class RPCClient {
     this.manager = new ClientManager(
       connection,
       this.options.parser,
-      this.options.subscribePolicy !== 'fail'
+      this.options.subscribePolicy !== 'fail',
+      this.options.batch
     );
   }
   public get status(): RPCClientStatus {

diff --git a/packages/rpc/src/client/defaults.ts b/packages/rpc/src/client/defaults.ts
@@ -3,6 +3,7 @@ import { DataInput, DataOutput } from '~/types';
 
 export function createDefaults(): Required<RPCClientOptions> {
   return {
+    batch: true,
     responseTimeout: 30000,
     subscribePolicy: 'fail',
     unsubscribePolicy: 'complete',

diff --git a/packages/rpc/src/client/types/client.ts b/packages/rpc/src/client/types/client.ts
@@ -4,6 +4,10 @@ import { DataParser } from '~/types';
 export type RPCClientStatus = RPCClientConnectionStatus | 'complete';
 
 export interface RPCClientOptions {
+  /**
+   * Whether to batch requests. Default: `true`.
+   */
+  batch?: boolean;
   /**
    * Timeout for request's first result when connected in milliseconds. `0` for *infinity.* It will cause all pending requests to fail. Default: `30000`.
    */