Add maximum time for flushing batches

Author: code-asher

packages/protocol/src/common/proxy.ts

@@ -92,20 +92,25 @@ interface BatchItem<T, A> {
  * Batch remote calls.
  */
 export abstract class Batch<T, A> {
-	/**
-	 * Flush after reaching this count.
-	 */
-	private maxCount = 100;
-
-	/**
-	 * Flush after not receiving more requests for this amount of time.
-	 */
-	private maxTime = 100;
-
-	private flushTimeout: number | NodeJS.Timer | undefined;
-
+	private idleTimeout: number | NodeJS.Timer | undefined;
+	private maxTimeout: number | NodeJS.Timer | undefined;
 	private batch = <BatchItem<T, A>[]>[];
 
+	public constructor(
+		/**
+		 * Flush after reaching this amount of time.
+		 */
+		private readonly maxTime = 1000,
+		/**
+		 * Flush after reaching this count.
+		 */
+		private readonly maxCount = 100,
+		/**
+		 * Flush after not receiving more requests for this amount of time.
+		 */
+		private readonly idleTime = 100,
+	) {}
+
 	public add = (args: A): Promise<T> => {
 		return new Promise((resolve, reject) => {
 			this.batch.push({
@@ -116,26 +121,29 @@ export abstract class Batch<T, A> {
 			if (this.batch.length >= this.maxCount) {
 				this.flush();
 			} else {
-				clearTimeout(this.flushTimeout as any);
-				this.flushTimeout = setTimeout(this.flush, this.maxTime);
+				clearTimeout(this.idleTimeout as any);
+				this.idleTimeout = setTimeout(this.flush, this.idleTime);
+				if (typeof this.maxTimeout === "undefined") {
+					this.maxTimeout = setTimeout(this.flush, this.maxTime);
+				}
 			}
 		});
 	}
 
 	protected abstract remoteCall(batch: A[]): Promise<(T | Error)[]>;
 
 	private flush = (): void => {
-		clearTimeout(this.flushTimeout as any);
+		clearTimeout(this.idleTimeout as any);
+		clearTimeout(this.maxTimeout as any);
+		this.maxTimeout = undefined;
 
 		const batch = this.batch;
 		this.batch = [];
 
 		this.remoteCall(batch.map((q) => q.args)).then((results) => {
 			batch.forEach((item, i) => {
 				const result = results[i];
-				if (!result) {
-					item.reject(new Error("no response"));
-				} else if (result instanceof Error) {
+				if (result && result instanceof Error) {
 					item.reject(result);
 				} else {
 					item.resolve(result);

packages/protocol/src/node/modules/fs.ts

@@ -157,7 +157,7 @@ export class FsModuleProxy {
 	}
 
 	public async lstatBatch(args: { path: fs.PathLike }[]): Promise<(Stats | Error)[]> {
-		return Promise.all(args.map((args) => this.lstat(args.path).catch((e) => e)));
+		return Promise.all(args.map((a) => this.lstat(a.path).catch((e) => e)));
 	}
 
 	public mkdir(path: fs.PathLike, mode: number | string | fs.MakeDirectoryOptions | undefined | null): Promise<void> {
@@ -187,7 +187,7 @@ export class FsModuleProxy {
 	}
 
 	public readdirBatch(args: { path: fs.PathLike, options: IEncodingOptions }[]): Promise<(Buffer[] | fs.Dirent[] | string[] | Error)[]> {
-		return Promise.all(args.map((args) => this.readdir(args.path, args.options).catch((e) => e)));
+		return Promise.all(args.map((a) => this.readdir(a.path, a.options).catch((e) => e)));
 	}
 
 	public readlink(path: fs.PathLike, options: IEncodingOptions): Promise<string | Buffer> {
@@ -211,7 +211,7 @@ export class FsModuleProxy {
 	}
 
 	public async statBatch(args: { path: fs.PathLike }[]): Promise<(Stats | Error)[]> {
-		return Promise.all(args.map((args) => this.stat(args.path).catch((e) => e)));
+		return Promise.all(args.map((a) => this.stat(a.path).catch((e) => e)));
 	}
 
 	public symlink(target: fs.PathLike, path: fs.PathLike, type?: fs.symlink.Type | null): Promise<void> {