Chore: AsyncActionQueue: Support changing which tasks can be skipped (#10506)

.eslintignore

@@ -788,6 +788,7 @@ packages/generator-joplin/generators/app/templates/src/index.js
 packages/generator-joplin/tools/updateCategories.js
 packages/htmlpack/src/index.js
 packages/lib/ArrayUtils.js
+packages/lib/AsyncActionQueue.test.js
 packages/lib/AsyncActionQueue.js
 packages/lib/BaseApplication.js
 packages/lib/BaseModel.js

.gitignore

@@ -767,6 +767,7 @@ packages/generator-joplin/generators/app/templates/src/index.js
 packages/generator-joplin/tools/updateCategories.js
 packages/htmlpack/src/index.js
 packages/lib/ArrayUtils.js
+packages/lib/AsyncActionQueue.test.js
 packages/lib/AsyncActionQueue.js
 packages/lib/BaseApplication.js
 packages/lib/BaseModel.js

packages/app-mobile/components/screens/Note.tsx

@@ -122,8 +122,7 @@ class NoteScreenComponent extends BaseScreenComponent<Props, State> implements B
 	// a re-render.
 	private lastBodyScroll: number|undefined = undefined;
 
-	// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Old code before rule was applied
+	private saveActionQueues_: Record<string, AsyncActionQueue>;
-	private saveActionQueues_: any;
 	private doFocusUpdate_: boolean;
 	// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Old code before rule was applied
 	private styles_: any;
@@ -595,7 +594,7 @@ class NoteScreenComponent extends BaseScreenComponent<Props, State> implements B
 
 		shared.uninstallResourceHandling(this.refreshResource);
 
-		this.saveActionQueue(this.state.note.id).processAllNow();
+		void this.saveActionQueue(this.state.note.id).processAllNow();
 
 		// It cannot theoretically be undefined, since componentDidMount should always be called before
 		// componentWillUnmount, but with React Native the impossible often becomes possible.

packages/lib/AsyncActionQueue.test.ts

@@ -0,0 +1,102 @@
+import AsyncActionQueue from './AsyncActionQueue';
+
+describe('AsyncActionQueue', () => {
+	beforeEach(() => {
+		jest.useRealTimers();
+	});
+
+	test('should run a single task', async () => {
+		const queue = new AsyncActionQueue(0);
+
+		await expect(new Promise(resolve => {
+			queue.push(() => resolve('success'));
+		})).resolves.toBe('success');
+
+		await expect(new Promise(resolve => {
+			queue.push(() => resolve('task 2'));
+		})).resolves.toBe('task 2');
+	});
+
+	test('should merge all tasks by default', async () => {
+		const queue = new AsyncActionQueue(100);
+		jest.useFakeTimers();
+
+		const result = {
+			ranFirst: false,
+			ranSecond: false,
+			ranThird: false,
+			ranFourth: false,
+		};
+		queue.push(() => {
+			result.ranFirst = true;
+		});
+		queue.push(() => {
+			result.ranSecond = true;
+		});
+		queue.push(() => {
+			result.ranThird = true;
+		});
+		queue.push(() => {
+			result.ranFourth = true;
+		});
+
+		const processPromise = queue.processAllNow();
+		await jest.runAllTimersAsync();
+		await processPromise;
+
+		expect(result).toMatchObject({
+			ranFirst: false,
+			ranSecond: false,
+			ranThird: false,
+			ranFourth: true,
+		});
+	});
+
+	test.each([
+		{
+			tasks: [
+				'a', 'b',
+			],
+			expectedToRun: [
+				'a', 'b',
+			],
+		},
+		{
+			tasks: [
+				'a', 'b', 'c',
+			],
+			expectedToRun: [
+				'a', 'b', 'c',
+			],
+		},
+		{
+			tasks: [
+				'group1', 'group1', 'group1', 'group2', 'group1', 'group1', 'group2', 'group2',
+			],
+			expectedToRun: [
+				'group1', 'group2', 'group1', 'group2',
+			],
+		},
+	])('should support customising how tasks are merged', async ({ tasks, expectedToRun }) => {
+		const queue = new AsyncActionQueue<string>(100);
+
+		// Determine which tasks can be merged based on their context
+		queue.setCanSkipTaskHandler((current, next) => {
+			return current.context === next.context;
+		});
+		jest.useFakeTimers();
+
+		const result: string[] = [];
+		for (const task of tasks) {
+			queue.push(() => {
+				result.push(task);
+			}, task);
+		}
+
+		const processPromise = queue.processAllNow();
+		await jest.runAllTimersAsync();
+		await processPromise;
+
+		expect(result).toMatchObject(expectedToRun);
+	});
+});

packages/lib/AsyncActionQueue.ts

@@ -1,13 +1,13 @@
+import Logger from '@joplin/utils/Logger';
 import shim from './shim';
 
-export interface QueueItemAction {
+export interface QueueItemAction<Context> {
-	(): void;
+	(context: Context): void|Promise<void>;
 }
 
-export interface QueueItem {
+export interface QueueItem<Context> {
-	action: QueueItemAction;
+	action: QueueItemAction<Context>;
-	// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Old code before rule was applied
+	context: Context;
-	context: any;
 }
 
 export enum IntervalType {
@@ -15,17 +15,20 @@ export enum IntervalType {
 	Fixed = 2,
 }
 
+type CanSkipTaskHandler<Context> = (current: QueueItem<Context>, next: QueueItem<Context>)=> boolean;
+
+const logger = Logger.create('AsyncActionQueue');
+
 // The AsyncActionQueue can be used to debounce asynchronous actions, to make sure
 // they run in the right order, and also to ensure that if multiple actions are emitted
 // only the last one is executed. This is particularly useful to save data in the background.
 // Each queue should be associated with a specific entity (a note, resource, etc.)
-export default class AsyncActionQueue {
+export default class AsyncActionQueue<Context = void> {
 
-	private queue_: QueueItem[] = [];
+	private queue_: QueueItem<Context>[] = [];
 	private interval_: number;
 	private intervalType_: number;
-	// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Old code before rule was applied
+	private scheduleProcessingIID_: ReturnType<typeof shim.setInterval>|null = null;
-	private scheduleProcessingIID_: any = null;
 	private processing_ = false;
 
 	private processingFinishedPromise_: Promise<void>;
@@ -43,8 +46,14 @@ export default class AsyncActionQueue {
 		});
 	}
 
-	// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Old code before rule was applied
+	// Determines whether an item can be skipped in the queue. Prevents data loss in the case that
-	public push(action: QueueItemAction, context: any = null) {
+	// tasks that do different things are added to the queue.
+	private canSkipTaskHandler_: CanSkipTaskHandler<Context> = (_current, _next) => true;
+	public setCanSkipTaskHandler(callback: CanSkipTaskHandler<Context>) {
+		this.canSkipTaskHandler_ = callback;
+	}
+
+	public push(action: QueueItemAction<Context>, context: Context = null) {
 		this.queue_.push({
 			action: action,
 			context: context,
@@ -76,17 +85,31 @@ export default class AsyncActionQueue {
 			return;
 		}
 
-		this.processing_ = true;
-
 		const itemCount = this.queue_.length;
 
 		if (itemCount) {
-			const item = this.queue_[itemCount - 1];
+			this.processing_ = true;
-			await item.action();
-			this.queue_.splice(0, itemCount);
-		}
 
-		this.processing_ = false;
+			let i = 0;
+			try {
+				for (i = 0; i < itemCount; i++) {
+					const current = this.queue_[i];
+					const next = i + 1 < itemCount ? this.queue_[i + 1] : null;
+					if (!next || !this.canSkipTaskHandler_(current, next)) {
+						await current.action(current.context);
+					}
+				}
+			} catch (error) {
+				i ++; // Don't repeat the failed task.
+				logger.warn('Unhandled error:', error);
+				throw error;
+			} finally {
+				// Removing processed items in a try {} finally {...} prevents
+				// items from being processed twice, even if one throws an Error.
+				this.queue_.splice(0, i);
+				this.processing_ = false;
+			}
+		}
 
 		if (this.queue_.length === 0) {
 			this.onProcessingFinished_();
@@ -114,4 +137,3 @@ export default class AsyncActionQueue {
 		return this.processingFinishedPromise_;
 	}
 }
-