Merge pull request #53584 from Microsoft/joao/remove-ppromise-chokidar

Remove ppromise usage from chokidar

src/vs/workbench/services/files/node/watcher/unix/chokidarWatcherService.ts

@@ -21,7 +21,8 @@ import { normalizeNFC } from 'vs/base/common/normalization';
 import { realcaseSync } from 'vs/base/node/extfs';
 import { isMacintosh } from 'vs/base/common/platform';
 import * as watcherCommon from 'vs/workbench/services/files/node/watcher/common';
-import { IWatcherRequest, IWatcherService, IWatcherOptions } from 'vs/workbench/services/files/node/watcher/unix/watcher';
+import { IWatcherRequest, IWatcherService, IWatcherOptions, IWatchError } from 'vs/workbench/services/files/node/watcher/unix/watcher';
+import { Emitter, Event } from 'vs/base/common/event';
 
 interface IWatcher {
 	requests: ExtendedWatcherRequest[];
@@ -44,29 +45,20 @@ export class ChokidarWatcherService implements IWatcherService {
 	private _watchers: { [watchPath: string]: IWatcher };
 	private _watcherCount: number;
 
-	private _watcherPromise: TPromise<void>;
 	private _options: IWatcherOptions & IChockidarWatcherOptions;
 
 	private spamCheckStartTime: number;
 	private spamWarningLogged: boolean;
 	private enospcErrorLogged: boolean;
-	private _errorCallback: (error: Error) => void;
-	private _fileChangeCallback: (changes: watcherCommon.IRawFileChange[]) => void;
 
-	public initialize(options: IWatcherOptions & IChockidarWatcherOptions): TPromise<void> {
+	private _onWatchEvent = new Emitter<watcherCommon.IRawFileChange[] | IWatchError>();
+	readonly onWatchEvent = this._onWatchEvent.event;
+
+	watch(options: IWatcherOptions & IChockidarWatcherOptions): Event<watcherCommon.IRawFileChange[] | IWatchError> {
 		this._options = options;
 		this._watchers = Object.create(null);
 		this._watcherCount = 0;
-		this._watcherPromise = new TPromise<void>((c, e, p) => {
-			this._errorCallback = (error) => {
-				this.stop();
-				e(error);
-			};
-			this._fileChangeCallback = p;
-		}, () => {
-			this.stop();
-		});
-		return this._watcherPromise;
+		return this.onWatchEvent;
 	}
 
 	public setRoots(requests: IWatcherRequest[]): TPromise<void> {
@@ -233,7 +225,7 @@ export class ChokidarWatcherService implements IWatcherService {
 
 				// Broadcast to clients normalized
 				const res = watcherCommon.normalize(events);
-				this._fileChangeCallback(res);
+				this._onWatchEvent.fire(res);
 
 				// Logging
 				if (this._options.verboseLogging) {
@@ -257,7 +249,8 @@ export class ChokidarWatcherService implements IWatcherService {
 				if ((<any>error).code === 'ENOSPC') {
 					if (!this.enospcErrorLogged) {
 						this.enospcErrorLogged = true;
-						this._errorCallback(new Error('Inotify limit reached (ENOSPC)'));
+						this.stop();
+						this._onWatchEvent.fire({ message: 'Inotify limit reached (ENOSPC)' });
 					}
 				} else {
 					console.error(error.toString());

src/vs/workbench/services/files/node/watcher/unix/test/chockidarWatcherService.test.ts

@@ -137,19 +137,15 @@ suite.skip('Chockidar watching', () => {
 		await pfs.mkdirp(bFolder);
 		await pfs.mkdirp(b2Folder);
 
-		const promise = service.initialize({ verboseLogging: false, pollingInterval: 200 });
-		promise.then(null,
-			e => {
-				console.log('set error', e);
-				error = e;
-			},
-			p => {
-				if (Array.isArray(p)) {
-					result.push(...p);
-				}
+		const opts = { verboseLogging: false, pollingInterval: 200 };
+		service.watch(opts)(e => {
+			if (Array.isArray(e)) {
+				result.push(...e);
+			} else {
+				console.log('set error', e.message);
+				error = e.message;
 			}
-		);
-
+		});
 	});
 
 	suiteTeardown(async () => {

src/vs/workbench/services/files/node/watcher/unix/watcher.ts

@@ -6,6 +6,8 @@
 'use strict';
 
 import { TPromise } from 'vs/base/common/winjs.base';
+import { Event } from 'vs/base/common/event';
+import { IRawFileChange } from 'vs/workbench/services/files/node/watcher/common';
 
 export interface IWatcherRequest {
 	basePath: string;
@@ -16,7 +18,11 @@ export interface IWatcherOptions {
 	verboseLogging: boolean;
 }
 
+export interface IWatchError {
+	message: string;
+}
+
 export interface IWatcherService {
-	initialize(options: IWatcherOptions): TPromise<void>;
+	watch(options: IWatcherOptions): Event<IRawFileChange[] | IWatchError>;
 	setRoots(roots: IWatcherRequest[]): TPromise<void>;
 }
\ No newline at end of file

src/vs/workbench/services/files/node/watcher/unix/watcherIpc.ts

@@ -7,26 +7,31 @@
 
 import { TPromise } from 'vs/base/common/winjs.base';
 import { IChannel } from 'vs/base/parts/ipc/common/ipc';
-import { IWatcherRequest, IWatcherService, IWatcherOptions } from 'vs/workbench/services/files/node/watcher/unix/watcher';
+import { IWatcherRequest, IWatcherService, IWatcherOptions, IWatchError } from './watcher';
 import { Event } from 'vs/base/common/event';
+import { IRawFileChange } from 'vs/workbench/services/files/node/watcher/common';
 
 export interface IWatcherChannel extends IChannel {
-	call(command: 'initialize', options: IWatcherOptions): TPromise<void>;
+	listen(event: 'watch', verboseLogging: boolean): Event<IRawFileChange[] | Error>;
+	listen<T>(event: string, arg?: any): Event<T>;
+
 	call(command: 'setRoots', request: IWatcherRequest[]): TPromise<void>;
-	call(command: string, arg: any): TPromise<any>;
+	call<T>(command: string, arg?: any): TPromise<T>;
 }
 
 export class WatcherChannel implements IWatcherChannel {
 
 	constructor(private service: IWatcherService) { }
 
-	listen<T>(event: string, arg?: any): Event<T> {
+	listen(event: string, arg?: any): Event<any> {
+		switch (event) {
+			case 'watch': return this.service.watch(arg);
+		}
 		throw new Error('No events');
 	}
 
 	call(command: string, arg: any): TPromise<any> {
 		switch (command) {
-			case 'initialize': return this.service.initialize(arg);
 			case 'setRoots': return this.service.setRoots(arg);
 		}
 		return undefined;
@@ -37,8 +42,8 @@ export class WatcherChannelClient implements IWatcherService {
 
 	constructor(private channel: IWatcherChannel) { }
 
-	initialize(options: IWatcherOptions): TPromise<void> {
-		return this.channel.call('initialize', options);
+	watch(options: IWatcherOptions): Event<IRawFileChange[] | IWatchError> {
+		return this.channel.listen('watch', options);
 	}
 
 	setRoots(roots: IWatcherRequest[]): TPromise<void> {

src/vs/workbench/services/files/node/watcher/unix/watcherService.ts

@@ -5,7 +5,6 @@
 
 'use strict';
 
-import { TPromise } from 'vs/base/common/winjs.base';
 import { getNextTickChannel } from 'vs/base/parts/ipc/common/ipc';
 import { Client } from 'vs/base/parts/ipc/node/ipc.cp';
 import uri from 'vs/base/common/uri';
@@ -13,10 +12,11 @@ import { toFileChangesEvent, IRawFileChange } from 'vs/workbench/services/files/
 import { IWatcherChannel, WatcherChannelClient } from 'vs/workbench/services/files/node/watcher/unix/watcherIpc';
 import { FileChangesEvent, IFilesConfiguration } from 'vs/platform/files/common/files';
 import { IWorkspaceContextService } from 'vs/platform/workspace/common/workspace';
-import { isPromiseCanceledError } from 'vs/base/common/errors';
 import { IDisposable, dispose } from 'vs/base/common/lifecycle';
 import { IConfigurationService } from 'vs/platform/configuration/common/configuration';
 import { Schemas } from 'vs/base/common/network';
+import { filterEvent } from 'vs/base/common/event';
+import { IWatchError } from 'vs/workbench/services/files/node/watcher/unix/watcher';
 
 export class FileWatcher {
 	private static readonly MAX_RESTARTS = 5;
@@ -55,20 +55,7 @@ export class FileWatcher {
 		);
 		this.toDispose.push(client);
 
-		const channel = getNextTickChannel(client.getChannel<IWatcherChannel>('watcher'));
-		this.service = new WatcherChannelClient(channel);
-
-		const options = {
-			verboseLogging: this.verboseLogging
-		};
-
-		this.service.initialize(options).then(null, err => {
-			if (!this.isDisposed && !isPromiseCanceledError(err)) {
-				return TPromise.wrapError(err); // the service lib uses the promise cancel error to indicate the process died, we do not want to bubble this up
-			}
-			return void 0;
-		}, (events: IRawFileChange[]) => this.onRawFileEvents(events)).done(() => {
-
+		client.onDidProcessExit(() => {
 			// our watcher app should never be completed because it keeps on watching. being in here indicates
 			// that the watcher process died and we want to restart it here. we only do it a max number of times
 			if (!this.isDisposed) {
@@ -80,11 +67,19 @@ export class FileWatcher {
 					this.errorLogger('[FileWatcher] failed to start after retrying for some time, giving up. Please report this as a bug report!');
 				}
 			}
-		}, error => {
-			if (!this.isDisposed) {
-				this.errorLogger(error);
-			}
-		});
+		}, null, this.toDispose);
+
+		const channel = getNextTickChannel(client.getChannel<IWatcherChannel>('watcher'));
+		this.service = new WatcherChannelClient(channel);
+
+		const options = { verboseLogging: this.verboseLogging };
+		const onWatchEvent = filterEvent(this.service.watch(options), () => !this.isDisposed);
+
+		const onError = filterEvent<any, IWatchError>(onWatchEvent, (e): e is IWatchError => typeof e.message === 'string');
+		onError(err => this.errorLogger(err.message), null, this.toDispose);
+
+		const onFileChanges = filterEvent<any, IRawFileChange[]>(onWatchEvent, (e): e is IRawFileChange[] => Array.isArray(e) && e.length > 0);
+		onFileChanges(e => this.onFileChanges(toFileChangesEvent(e)), null, this.toDispose);
 
 		// Start watching
 		this.updateFolders();
@@ -123,17 +118,6 @@ export class FileWatcher {
 		}));
 	}
 
-	private onRawFileEvents(events: IRawFileChange[]): void {
-		if (this.isDisposed) {
-			return;
-		}
-
-		// Emit through event emitter
-		if (events.length > 0) {
-			this.onFileChanges(toFileChangesEvent(events));
-		}
-	}
-
 	private dispose(): void {
 		this.isDisposed = true;
 		this.toDispose = dispose(this.toDispose);