From 94d67cef71253d591512f547e4b265df4c9e79ea Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Sun, 3 Dec 2023 00:01:23 +0100
Subject: [PATCH] Add AsyncSerialCallback helper

Add helper to serialise the calls made to an async function such that no
overlapping invocations happen to the async function by queue up calls
made while it's running to continue after it's running.
---
 packages/lib/src/helpers.ts | 48 ++++++++++++++++++++
 test/lib/helpers.js         | 88 +++++++++++++++++++++++++++++++++++++
 2 files changed, 136 insertions(+)

diff --git a/packages/lib/src/helpers.ts b/packages/lib/src/helpers.ts
index 38c804b3..c9589d92 100644
--- a/packages/lib/src/helpers.ts
+++ b/packages/lib/src/helpers.ts
@@ -107,6 +107,54 @@ export class AsyncSerialMergingCallback {
 	}
 }
 
+/**
+ * Helper to serialise calls to an async callback.
+ *
+ * Provides a convenient interface to serialise the calls made to an async
+ * callback such that no overlapping invocations of the callback is made,
+ * and successive calls are served on a first in first out basis.  If the
+ * callback is already running when another invocation is requested the
+ * request will be queued until the callback returns and then the callback
+ * called again.
+ */
+export class AsyncSerialCallback<
+	Callback extends (...args: Parameters<Callback>) => Promise<Awaited<ReturnType<Callback>>>,
+> {
+	private _currentlyRunning = false;
+	private _currentlyWaiting: (() => void)[] = [];
+
+	/**
+	 * @param callback - Async function to serialise access to
+	 */
+	constructor(
+		public callback: Callback,
+	) { }
+
+	/**
+	 * Invoke the assosiated callback.
+	 *
+	 * If the callback is currently running then this will wait until the
+	 * existing invocation finishes and then invoke it again. If called
+	 * multiple times while an invocation is running the calls will be
+	 * merged into one call.
+	 */
+	async invoke(...args: Parameters<Callback>) {
+		if (this._currentlyRunning) {
+			await new Promise<void>(resolve => {
+				this._currentlyWaiting.push(resolve);
+			});
+		}
+
+		this._currentlyRunning = true;
+		try {
+			return await this.callback(...args);
+		} finally {
+			this._currentlyRunning = false;
+			this._currentlyWaiting.shift()?.();
+		}
+	}
+}
+
 
 /**
  * Read stream to the end and return its content
diff --git a/test/lib/helpers.js b/test/lib/helpers.js
index f9466d75..25b0270c 100644
--- a/test/lib/helpers.js
+++ b/test/lib/helpers.js
@@ -104,6 +104,94 @@ describe("lib/helpers", function() {
 		});
 	});
 
+	describe("class AsyncSerialCallback", function() {
+		let started;
+		let ended;
+		let serialiser;
+		beforeEach(function() {
+			started = 0;
+			ended = 0;
+			serialiser = new lib.AsyncSerialCallback(
+				(input) => new Promise((resolve, reject) => {
+					started += 1;
+					process.nextTick(() => {
+						ended += 1;
+						if (input !== 99) {
+							resolve(input);
+						} else {
+							reject(new Error("99"));
+						}
+					});
+				})
+			);
+		});
+		it("should await the callback when invoked", async function() {
+			await serialiser.invoke();
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 1, "incorrect ended count");
+		});
+		it("should return the result of the callback", async function() {
+			const id = await serialiser.invoke(42);
+			assert.equal(id, 42, "returned result not passed");
+		});
+		it("should immediately call the callback", async function() {
+			let promise = serialiser.invoke();
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 0, "incorrect ended count");
+			await promise;
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 1, "incorrect ended count");
+		});
+		it("should call the callback twice if invoked in twice in paralell", async function() {
+			await Promise.all([
+				serialiser.invoke(),
+				serialiser.invoke(),
+			]);
+			assert.equal(started, 2, "incorrect started count");
+			assert.equal(ended, 2, "incorrect ended count");
+		});
+		it("should call the callback many if invoked many times in paralell", async function() {
+			await Promise.all([
+				serialiser.invoke(),
+				serialiser.invoke(),
+				serialiser.invoke(),
+				serialiser.invoke(),
+			]);
+			assert.equal(started, 4, "incorrect started count");
+			assert.equal(ended, 4, "incorrect ended count");
+		});
+		it("should serialise calls", async function() {
+			let events = [];
+			let calls = await Promise.all([
+				serialiser.invoke(11).then((res) => { events.push(1); return res; }),
+				serialiser.invoke(12).then((res) => { events.push(2); return res; }),
+				serialiser.invoke(13).then((res) => { events.push(3); return res; }),
+				serialiser.invoke(14).then((res) => { events.push(4); return res; }),
+			]);
+			assert.deepEqual(events, [1, 2, 3, 4]);
+			assert.deepEqual(calls, [11, 12, 13, 14]);
+		});
+		it("should forward rejections", async function() {
+			await assert.rejects(serialiser.invoke(99), new Error("99"));
+		});
+		it("should handle queued rejections", async function() {
+			let events = [];
+			let calls = await Promise.allSettled([
+				serialiser.invoke(11).then((res) => { events.push(1); return res; }),
+				serialiser.invoke(99).then((res) => { events.push(2); return res; }),
+				serialiser.invoke(13).then((res) => { events.push(3); return res; }),
+				serialiser.invoke(14).then((res) => { events.push(4); return res; }),
+			]);
+			assert.deepEqual(events, [1, 3, 4]);
+			assert.deepEqual(calls, [
+				{ status: "fulfilled", value: 11 },
+				{ status: "rejected", reason: new Error("99") },
+				{ status: "fulfilled", value: 13 },
+				{ status: "fulfilled", value: 14 },
+			]);
+		});
+	});
+
 	function parse(input, attributes) {
 		const result = lib.parseSearchString(input, attributes);
 		if (!result.issues.length) {