Implemented a BufferedQueue, and refactored to use it for RepoManager File System Event processing. Various unit testing improvements.

Author: mhutchie

src/repoManager.ts

@@ -6,6 +6,7 @@ import { DEFAULT_REPO_STATE, ExtensionState } from './extensionState';
 import { Logger } from './logger';
 import { GitRepoSet, GitRepoState } from './types';
 import { evalPromises, getPathFromUri, pathWithTrailingSlash, realpath } from './utils';
+import { BufferedQueue } from './utils/bufferedQueue';
 import { Disposable, toDisposable } from './utils/disposable';
 import { Event, EventEmitter } from './utils/event';
 
@@ -22,18 +23,16 @@ export class RepoManager extends Disposable {
 	private readonly dataSource: DataSource;
 	private readonly extensionState: ExtensionState;
 	private readonly logger: Logger;
+
 	private repos: GitRepoSet;
 	private ignoredRepos: string[];
 	private maxDepthOfRepoSearch: number;
+
 	private readonly folderWatchers: { [workspace: string]: vscode.FileSystemWatcher } = {};
 	private readonly repoEventEmitter: EventEmitter<RepoChangeEvent>;
 
-	private readonly createEventQueue: string[] = [];
-	private readonly changeEventQueue: string[] = [];
-	private processCreateEventsTimeout: NodeJS.Timer | null = null;
-	private processChangeEventsTimeout: NodeJS.Timer | null = null;
-	private processingCreateEvents: boolean = false;
-	private processingChangeEvents: boolean = false;
+	private readonly onWatcherCreateQueue: BufferedQueue<string>;
+	private readonly onWatcherChangeQueue: BufferedQueue<string>;
 
 	/**
 	 * Creates the Git Graph Repository Manager, and runs startup tasks.
@@ -49,7 +48,12 @@ export class RepoManager extends Disposable {
 		this.repos = extensionState.getRepos();
 		this.ignoredRepos = extensionState.getIgnoredRepos();
 		this.maxDepthOfRepoSearch = getConfig().maxDepthOfRepoSearch;
+
 		this.repoEventEmitter = new EventEmitter<RepoChangeEvent>();
+
+		this.onWatcherCreateQueue = new BufferedQueue<string>(this.processOnWatcherCreateEvent.bind(this), this.sendRepos.bind(this));
+		this.onWatcherChangeQueue = new BufferedQueue<string>(this.processOnWatcherChangeEvent.bind(this), this.sendRepos.bind(this));
+
 		this.startupTasks();
 
 		this.registerDisposables(
@@ -83,6 +87,12 @@ export class RepoManager extends Disposable {
 			// Dispose the Repository Event Emitter when disposed
 			this.repoEventEmitter,
 
+			// Dispose the onWatcherCreateQueue
+			this.onWatcherCreateQueue,
+
+			// Dispose the onWatcherChangeQueue
+			this.onWatcherChangeQueue,
+
 			// Stop watching folders when disposed
 			toDisposable(() => {
 				const folders = Object.keys(this.folderWatchers);
@@ -483,9 +493,9 @@ export class RepoManager extends Disposable {
 	 */
 	private startWatchingFolder(path: string) {
 		let watcher = vscode.workspace.createFileSystemWatcher(path + '/**');
-		watcher.onDidCreate(uri => this.onWatcherCreate(uri));
-		watcher.onDidChange(uri => this.onWatcherChange(uri));
-		watcher.onDidDelete(uri => this.onWatcherDelete(uri));
+		watcher.onDidCreate((uri) => this.onWatcherCreate(uri));
+		watcher.onDidChange((uri) => this.onWatcherChange(uri));
+		watcher.onDidDelete((uri) => this.onWatcherDelete(uri));
 		this.folderWatchers[path] = watcher;
 	}
 
@@ -499,53 +509,29 @@ export class RepoManager extends Disposable {
 	}
 
 	/**
-	 * Process a file system creation event.
+	 * Handle a file system creation event.
 	 * @param uri The URI of the creation event.
 	 */
 	private onWatcherCreate(uri: vscode.Uri) {
 		let path = getPathFromUri(uri);
 		if (path.indexOf('/.git/') > -1) return;
 		if (path.endsWith('/.git')) path = path.slice(0, -5);
-		if (this.createEventQueue.indexOf(path) > -1) return;
-
-		this.createEventQueue.push(path);
-
-		if (!this.processingCreateEvents) {
-			if (this.processCreateEventsTimeout !== null) {
-				clearTimeout(this.processCreateEventsTimeout);
-			}
-			this.processCreateEventsTimeout = setTimeout(() => {
-				this.processCreateEventsTimeout = null;
-				this.processCreateEvents();
-			}, 1000);
-		}
+		this.onWatcherCreateQueue.enqueue(path);
 	}
 
 	/**
-	 * Process a file system change event.
+	 * Handle a file system change event.
 	 * @param uri The URI of the change event.
 	 */
 	private onWatcherChange(uri: vscode.Uri) {
 		let path = getPathFromUri(uri);
 		if (path.indexOf('/.git/') > -1) return;
 		if (path.endsWith('/.git')) path = path.slice(0, -5);
-		if (this.changeEventQueue.indexOf(path) > -1) return;
-
-		this.changeEventQueue.push(path);
-
-		if (!this.processingChangeEvents) {
-			if (this.processChangeEventsTimeout !== null) {
-				clearTimeout(this.processChangeEventsTimeout);
-			}
-			this.processChangeEventsTimeout = setTimeout(() => {
-				this.processChangeEventsTimeout = null;
-				this.processChangeEvents();
-			}, 1000);
-		}
+		this.onWatcherChangeQueue.enqueue(path);
 	}
 
 	/**
-	 * Process a file system deletion event.
+	 * Handle a file system deletion event.
 	 * @param uri The URI of the deletion event.
 	 */
 	private onWatcherDelete(uri: vscode.Uri) {
@@ -556,33 +542,31 @@ export class RepoManager extends Disposable {
 	}
 
 	/**
-	 * Process the queue of file system creation events.
+	 * Process a file system creation event.
+	 * @param path The path of the file that was created.
+	 * @returns TRUE => Change was made. FALSE => No change was made.
 	 */
-	private async processCreateEvents() {
-		this.processingCreateEvents = true;
-		let path, changes = false;
-		while (path = this.createEventQueue.shift()) {
-			if (await isDirectory(path)) {
-				if (await this.searchDirectoryForRepos(path, this.maxDepthOfRepoSearch)) changes = true;
+	private async processOnWatcherCreateEvent(path: string) {
+		if (await isDirectory(path)) {
+			if (await this.searchDirectoryForRepos(path, this.maxDepthOfRepoSearch)) {
+				return true;
 			}
 		}
-		this.processingCreateEvents = false;
-		if (changes) this.sendRepos();
+		return false;
 	}
 
 	/**
-	 * Process the queue of file system change events
+	 * Process a file system change event.
+	 * @param path The path of the file that was changed.
+	 * @returns TRUE => Change was made. FALSE => No change was made.
 	 */
-	private async processChangeEvents() {
-		this.processingChangeEvents = true;
-		let path, changes = false;
-		while (path = this.changeEventQueue.shift()) {
-			if (!await doesPathExist(path)) {
-				if (this.removeReposWithinFolder(path)) changes = true;
+	private async processOnWatcherChangeEvent(path: string) {
+		if (!await doesPathExist(path)) {
+			if (this.removeReposWithinFolder(path)) {
+				return true;
 			}
 		}
-		this.processingChangeEvents = false;
-		if (changes) this.sendRepos();
+		return false;
 	}
 }
 

src/utils/bufferedQueue.ts

@@ -0,0 +1,72 @@
+import { Disposable, toDisposable } from './disposable';
+
+/**
+ * Represents a BufferedQueue, which is queue that buffers items for a short period of time before processing them.
+ */
+export class BufferedQueue<T> extends Disposable {
+	private readonly queue: T[] = [];
+	private timeout: NodeJS.Timer | null = null;
+	private processing: boolean = false;
+
+	private readonly bufferDuration: number;
+	private onItem: (item: T) => Promise<boolean>;
+	private onChanges: () => void;
+
+	/**
+	 * Constructs a BufferedQueue instance.
+	 * @param onItem A callback invoked to process an item in the queue.
+	 * @param onChanges A callback invoked when a change was indicated by onItem.
+	 * @param bufferDuration The number of milliseconds to buffer items in the queue.
+	 * @returns The BufferedQueue instance.
+	 */
+	constructor(onItem: (item: T) => Promise<boolean>, onChanges: () => void, bufferDuration: number = 1000) {
+		super();
+		this.bufferDuration = bufferDuration;
+		this.onItem = onItem;
+		this.onChanges = onChanges;
+
+		this.registerDisposable(toDisposable(() => {
+			if (this.timeout !== null) {
+				clearTimeout(this.timeout);
+				this.timeout = null;
+			}
+		}));
+	}
+
+	/**
+	 * Enqueue an item if it doesn't already exist in the queue.
+	 * @param item The item to enqueue.
+	 */
+	public enqueue(item: T) {
+		const itemIndex = this.queue.indexOf(item);
+		if (itemIndex > -1) {
+			this.queue.splice(itemIndex, 1);
+		}
+		this.queue.push(item);
+
+		if (!this.processing) {
+			if (this.timeout !== null) {
+				clearTimeout(this.timeout);
+			}
+			this.timeout = setTimeout(() => {
+				this.timeout = null;
+				this.run();
+			}, this.bufferDuration);
+		}
+	}
+
+	/**
+	 * Process all of the items that are currently queued, and call the onChanges callback if any of the items resulted in a change
+	 */
+	private async run() {
+		this.processing = true;
+		let item, changes = false;
+		while (item = this.queue.shift()) {
+			if (await this.onItem(item)) {
+				changes = true;
+			}
+		}
+		this.processing = false;
+		if (changes) this.onChanges();
+	}
+}

tests/avatarManager.test.ts

@@ -1,3 +1,5 @@
+import { waitForExpect } from './helpers/expectations';
+
 import * as date from './mocks/date';
 import * as vscode from './mocks/vscode';
 jest.mock('vscode', () => vscode, { virtual: true });
@@ -1737,24 +1739,6 @@ function expectFileToHaveBeenRead(name: string) {
 	expect(spyOnReadFile.mock.calls.some((args) => args[0] === name)).toBe(true);
 }
 
-function waitForExpect(expect: () => void) {
-	return new Promise((resolve, reject) => {
-		let attempts = 0;
-		const testInterval = setInterval(async () => {
-			try {
-				attempts++;
-				expect();
-				resolve();
-			} catch (e) {
-				if (attempts === 100) {
-					clearInterval(testInterval);
-					reject(e);
-				}
-			}
-		}, 50);
-	});
-}
-
 function waitForEvents(avatarManager: AvatarManager, n: number, runPendingTimers = false) {
 	return new Promise<AvatarEvent[]>((resolve) => {
 		const events: AvatarEvent[] = [];