Implement serial calls to save with a helper

Separate out the serialisation of the invocations to saveData with an
AsyncSerialMergingCallback helper than has unit tests covering its
functionality.

packages/controller/src/Controller.ts

@@ -376,9 +376,6 @@ export default class Controller {
 		}
 	}
 
-	private _saveDataCurrentlyRunning = false;
-	private _saveDataCurrentlyWaiting: (() => void)[] = [];
-
 	/**
 	 * Save all data currently in memory to disk
 	 *
@@ -386,31 +383,18 @@ export default class Controller {
 	 * completes and start another save.
 	 */
 	async saveData() {
-		if (this._saveDataCurrentlyRunning) {
-			const waitForCompletion = () => new Promise<void>(resolve => {
-				this._saveDataCurrentlyWaiting.push(resolve);
-			});
-			await waitForCompletion();
-			if (this._saveDataCurrentlyRunning) {
-				await waitForCompletion();
-				return;
-			}
-		}
+		await this._saveDataAsyncSerial.invoke();
+	}
 
-		this._saveDataCurrentlyRunning = true;
+	private _saveDataAsyncSerial = new lib.AsyncSerialMergingCallback(async () => {
 		try {
 			await this._saveDataInternal();
 		} catch (err: any) {
 			logger.error(`Unexpected error during saveData:\n${err.stack}`);
 		}
-		this._saveDataCurrentlyRunning = false;
-		for (const callback of this._saveDataCurrentlyWaiting) {
-			callback();
-		}
-		this._saveDataCurrentlyWaiting.length = 0;
-	}
+	});
 
-	async _saveDataInternal() {
+	private async _saveDataInternal() {
 		if (this.config.dirty) {
 			this.config.dirty = false;
 			await lib.safeOutputFile(this.configPath, JSON.stringify(this.config.serialize(), null, "\t"));

packages/lib/src/helpers.ts

@@ -51,6 +51,62 @@ export async function timeout<T>(promise: Promise<T>, time: number, timeoutResul
 	}
 }
 
+/**
+ * Helper to serialise and merge waiting 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.
+ * If the callback is already running when another invocation is requested
+ * the request will be queued until the callback returns and then an the
+ * callback called again.
+ *
+ * If multiple calls are made while the callback is running then these will
+ * be merged into one call of the callback.
+ */
+export class AsyncSerialMergingCallback {
+	private _currentlyRunning = false;
+	private _currentlyWaiting: (() => void)[] = [];
+
+	/**
+	 * @param callback - Async function to serialise access to
+	 */
+	constructor(
+		public callback: () => Promise<void>,
+	) { }
+
+	/**
+	 * 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() {
+		if (this._currentlyRunning) {
+			const waitForCompletion = () => new Promise<void>(resolve => {
+				this._currentlyWaiting.push(resolve);
+			});
+			await waitForCompletion();
+			if (this._currentlyRunning) {
+				await waitForCompletion();
+				return;
+			}
+		}
+
+		this._currentlyRunning = true;
+		try {
+			await this.callback();
+		} finally {
+			this._currentlyRunning = false;
+			for (const waiter of this._currentlyWaiting) {
+				waiter();
+			}
+			this._currentlyWaiting.length = 0;
+		}
+	}
+}
+
 
 /**
  * Read stream to the end and return its content

test/lib/helpers.js

@@ -44,6 +44,66 @@ describe("lib/helpers", function() {
 		});
 	});
 
+	describe("class AsyncSerialMergingCallback", function() {
+		let started;
+		let ended;
+		let merger;
+		beforeEach(function() {
+			started = 0;
+			ended = 0;
+			merger = new lib.AsyncSerialMergingCallback(
+				() => new Promise(resolve => {
+					started += 1;
+					process.nextTick(() => {
+						ended += 1;
+						resolve();
+					});
+				})
+			);
+		});
+		it("should await the callback when invoked", async function() {
+			await merger.invoke();
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 1, "incorrect ended count");
+		});
+		it("should immediately call the callback", async function() {
+			let promise = merger.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([
+				merger.invoke(),
+				merger.invoke(),
+			]);
+			assert.equal(started, 2, "incorrect started count");
+			assert.equal(ended, 2, "incorrect ended count");
+		});
+		it("should call the callback twice if invoked many times in paralell", async function() {
+			await Promise.all([
+				merger.invoke(),
+				merger.invoke(),
+				merger.invoke(),
+				merger.invoke(),
+			]);
+			assert.equal(started, 2, "incorrect started count");
+			assert.equal(ended, 2, "incorrect ended count");
+		});
+		it("should serialise calls", async function() {
+			let events = [];
+			await Promise.all([
+				merger.invoke().then(() => { events.push(1); }),
+				merger.invoke().then(() => { events.push(2); }),
+				merger.invoke().then(() => { events.push(3); }),
+				merger.invoke().then(() => { events.push(4); }),
+			]);
+			assert.deepEqual(events, [1, 2, 3, 4]);
+		});
+	});
+
 	function parse(input, attributes) {
 		const result = lib.parseSearchString(input, attributes);
 		if (!result.issues.length) {