More Observables (#291)

Author: felixfbecker

src/fs.ts

@@ -1,29 +1,29 @@
+import { Observable } from '@reactivex/rxjs';
+import { Glob } from 'glob';
 import * as fs from 'mz/fs';
-import { LanguageClient } from './lang-handler';
-import glob = require('glob');
-import iterate from 'iterare';
 import { Span } from 'opentracing';
 import Semaphore from 'semaphore-async-await';
+import { LanguageClient } from './lang-handler';
 import { InMemoryFileSystem } from './memfs';
-import { tracePromise } from './tracing';
+import { traceObservable } from './tracing';
 import { normalizeUri, uri2path } from './util';
 
 export interface FileSystem {
 	/**
 	 * Returns all files in the workspace under base
 	 *
 	 * @param base A URI under which to search, resolved relative to the rootUri
-	 * @return A promise that is fulfilled with an array of URIs
+	 * @return An Observable that emits URIs
 	 */
-	getWorkspaceFiles(base?: string, childOf?: Span): Promise<Iterable<string>>;
+	getWorkspaceFiles(base?: string, childOf?: Span): Observable<string>;
 
 	/**
 	 * Returns the content of a text document
 	 *
 	 * @param uri The URI of the text document, resolved relative to the rootUri
-	 * @return A promise that is fulfilled with the text document content
+	 * @return An Observable that emits the text document content
 	 */
-	getTextDocumentContent(uri: string, childOf?: Span): Promise<string>;
+	getTextDocumentContent(uri: string, childOf?: Span): Observable<string>;
 }
 
 export class RemoteFileSystem implements FileSystem {
@@ -34,17 +34,18 @@ export class RemoteFileSystem implements FileSystem {
 	 * The files request is sent from the server to the client to request a list of all files in the workspace or inside the directory of the base parameter, if given.
 	 * A language server can use the result to index files by filtering and doing a content request for each text document of interest.
 	 */
-	async getWorkspaceFiles(base?: string, childOf = new Span()): Promise<Iterable<string>> {
-		return iterate(await this.client.workspaceXfiles({ base }, childOf))
+	getWorkspaceFiles(base?: string, childOf = new Span()): Observable<string> {
+		return this.client.workspaceXfiles({ base }, childOf)
+			.mergeMap(textDocuments => textDocuments)
 			.map(textDocument => normalizeUri(textDocument.uri));
 	}
 
 	/**
 	 * The content request is sent from the server to the client to request the current content of any text document. This allows language servers to operate without accessing the file system directly.
 	 */
-	async getTextDocumentContent(uri: string, childOf = new Span()): Promise<string> {
-		const textDocument = await this.client.textDocumentXcontent({ textDocument: { uri } }, childOf);
-		return textDocument.text;
+	getTextDocumentContent(uri: string, childOf = new Span()): Observable<string> {
+		return this.client.textDocumentXcontent({ textDocument: { uri } }, childOf)
+			.map(textDocument => textDocument.text);
 	}
 }
 
@@ -62,24 +63,36 @@ export class LocalFileSystem implements FileSystem {
 		return uri2path(uri);
 	}
 
-	async getWorkspaceFiles(base = this.rootUri): Promise<Iterable<string>> {
+	getWorkspaceFiles(base = this.rootUri): Observable<string> {
 		if (!base.endsWith('/')) {
 			base += '/';
 		}
 		const cwd = this.resolveUriToPath(base);
-		const files = await new Promise<string[]>((resolve, reject) => {
-			glob('*', {
+		return new Observable<string>(subscriber => {
+			const globber = new Glob('*', {
 				cwd,
 				nodir: true,
 				matchBase: true,
 				follow: true
-			}, (err, matches) => err ? reject(err) : resolve(matches));
+			});
+			globber.on('match', (file: string) => {
+				subscriber.next(normalizeUri(base + file));
+			});
+			globber.on('error', (err: any) => {
+				subscriber.error(err);
+			});
+			globber.on('end', () => {
+				subscriber.complete();
+			});
+			return () => {
+				globber.abort();
+			};
 		});
-		return iterate(files).map(file => normalizeUri(base + file));
 	}
 
-	async getTextDocumentContent(uri: string): Promise<string> {
-		return fs.readFile(this.resolveUriToPath(uri), 'utf8');
+	getTextDocumentContent(uri: string): Observable<string> {
+		const filePath = this.resolveUriToPath(uri);
+		return Observable.fromPromise(fs.readFile(filePath, 'utf8'));
 	}
 }
 
@@ -91,14 +104,14 @@ export class LocalFileSystem implements FileSystem {
 export class FileSystemUpdater {
 
 	/**
-	 * Promise for a pending or fulfilled structure fetch
+	 * Observable for a pending or completed structure fetch
 	 */
-	private structureFetch?: Promise<void>;
+	private structureFetch?: Observable<never>;
 
 	/**
-	 * Map from URI to Promise of pending or fulfilled content fetch
+	 * Map from URI to Observable of pending or completed content fetch
 	 */
-	private fetches = new Map<string, Promise<void>>();
+	private fetches = new Map<string, Observable<never>>();
 
 	/**
 	 * Limits concurrent fetches to not fetch thousands of files in parallel
@@ -112,20 +125,23 @@ export class FileSystemUpdater {
 	 *
 	 * @param uri URI of the file to fetch
 	 * @param childOf A parent span for tracing
+	 * @return Observable that completes when the fetch is finished
 	 */
-	async fetch(uri: string, childOf = new Span()): Promise<void> {
+	fetch(uri: string, childOf = new Span()): Observable<never> {
 		// Limit concurrent fetches
-		const promise = this.concurrencyLimit.execute(async () => {
-			try {
-				const content = await this.remoteFs.getTextDocumentContent(uri);
+		const observable = Observable.fromPromise(this.concurrencyLimit.wait())
+			.mergeMap(() => this.remoteFs.getTextDocumentContent(uri))
+			.do(content => {
+				this.concurrencyLimit.signal();
 				this.inMemoryFs.add(uri, content);
-			} catch (err) {
+			}, err => {
 				this.fetches.delete(uri);
-				throw err;
-			}
-		});
-		this.fetches.set(uri, promise);
-		return promise;
+			})
+			.ignoreElements()
+			.publishReplay()
+			.refCount();
+		this.fetches.set(uri, observable);
+		return observable;
 	}
 
 	/**
@@ -134,9 +150,10 @@ export class FileSystemUpdater {
 	 *
 	 * @param uri URI of the file to ensure
 	 * @param childOf An OpenTracing span for tracing
+	 * @return Observable that completes when the file was fetched
 	 */
-	ensure(uri: string, childOf = new Span()): Promise<void> {
-		return tracePromise('Ensure content', childOf, span => {
+	ensure(uri: string, childOf = new Span()): Observable<never> {
+		return traceObservable('Ensure content', childOf, span => {
 			span.addTags({ uri });
 			return this.fetches.get(uri) || this.fetch(uri, span);
 		});
@@ -147,20 +164,20 @@ export class FileSystemUpdater {
 	 *
 	 * @param childOf A parent span for tracing
 	 */
-	fetchStructure(childOf = new Span()): Promise<void> {
-		const promise = tracePromise('Fetch workspace structure', childOf, async span => {
-			try {
-				const uris = await this.remoteFs.getWorkspaceFiles(undefined, span);
-				for (const uri of uris) {
+	fetchStructure(childOf = new Span()): Observable<never> {
+		const observable = traceObservable('Fetch workspace structure', childOf, span =>
+			this.remoteFs.getWorkspaceFiles(undefined, span)
+				.do(uri => {
 					this.inMemoryFs.add(uri);
-				}
-			} catch (err) {
-				this.structureFetch = undefined;
-				throw err;
-			}
-		});
-		this.structureFetch = promise;
-		return promise;
+				}, err => {
+					this.structureFetch = undefined;
+				})
+				.ignoreElements()
+				.publishReplay()
+				.refCount()
+		);
+		this.structureFetch = observable;
+		return observable;
 	}
 
 	/**
@@ -169,8 +186,8 @@ export class FileSystemUpdater {
 	 *
 	 * @param span An OpenTracing span for tracing
 	 */
-	ensureStructure(childOf = new Span()) {
-		return tracePromise('Ensure structure', childOf, span => {
+	ensureStructure(childOf = new Span()): Observable<never> {
+		return traceObservable('Ensure structure', childOf, span => {
 			return this.structureFetch || this.fetchStructure(span);
 		});
 	}

src/lang-handler.ts

@@ -26,13 +26,13 @@ export interface LanguageClient {
 	 * any text document. This allows language servers to operate without accessing the file system
 	 * directly.
 	 */
-	textDocumentXcontent(params: TextDocumentContentParams, childOf?: Span): Promise<TextDocumentItem>;
+	textDocumentXcontent(params: TextDocumentContentParams, childOf?: Span): Observable<TextDocumentItem>;
 
 	/**
 	 * The files request is sent from the server to the client to request a list of all files in the
 	 * workspace or inside the directory of the `base` parameter, if given.
 	 */
-	workspaceXfiles(params: WorkspaceFilesParams, childOf?: Span): Promise<TextDocumentIdentifier[]>;
+	workspaceXfiles(params: WorkspaceFilesParams, childOf?: Span): Observable<TextDocumentIdentifier[]>;
 
 	/**
 	 * The log message notification is sent from the server to the client to ask
@@ -44,7 +44,7 @@ export interface LanguageClient {
 	 * The cache get request is sent from the server to the client to request the value of a cache
 	 * item identified by the provided key.
 	 */
-	xcacheGet(params: CacheGetParams, childOf?: Span): Promise<any>;
+	xcacheGet(params: CacheGetParams, childOf?: Span): Observable<any>;
 
 	/**
 	 * The cache set notification is sent from the server to the client to set the value of a cache
@@ -65,7 +65,7 @@ export interface LanguageClient {
 	 * Can occur as as a result of rename or executeCommand (code action).
 	 * @param params The edits to apply to the workspace
 	 */
-	workspaceApplyEdit(params: ApplyWorkspaceEditParams, childOf?: Span): Promise<ApplyWorkspaceEditResponse>;
+	workspaceApplyEdit(params: ApplyWorkspaceEditParams, childOf?: Span): Observable<ApplyWorkspaceEditResponse>;
 }
 
 /**
@@ -145,16 +145,16 @@ export class RemoteLanguageClient {
 	 * any text document. This allows language servers to operate without accessing the file system
 	 * directly.
 	 */
-	textDocumentXcontent(params: TextDocumentContentParams, childOf = new Span()): Promise<TextDocumentItem> {
-		return this.request('textDocument/xcontent', params, childOf).toPromise();
+	textDocumentXcontent(params: TextDocumentContentParams, childOf = new Span()): Observable<TextDocumentItem> {
+		return this.request('textDocument/xcontent', params, childOf);
 	}
 
 	/**
 	 * The files request is sent from the server to the client to request a list of all files in the
 	 * workspace or inside the directory of the `base` parameter, if given.
 	 */
-	workspaceXfiles(params: WorkspaceFilesParams, childOf = new Span()): Promise<TextDocumentIdentifier[]> {
-		return this.request('workspace/xfiles', params, childOf).toPromise();
+	workspaceXfiles(params: WorkspaceFilesParams, childOf = new Span()): Observable<TextDocumentIdentifier[]> {
+		return this.request('workspace/xfiles', params, childOf);
 	}
 
 	/**
@@ -169,8 +169,8 @@ export class RemoteLanguageClient {
 	 * The cache get request is sent from the server to the client to request the value of a cache
 	 * item identified by the provided key.
 	 */
-	xcacheGet(params: CacheGetParams, childOf = new Span()): Promise<any> {
-		return this.request('xcache/get', params, childOf).toPromise();
+	xcacheGet(params: CacheGetParams, childOf = new Span()): Observable<any> {
+		return this.request('xcache/get', params, childOf);
 	}
 
 	/**
@@ -179,7 +179,7 @@ export class RemoteLanguageClient {
 	 * because the server is not supposed to act differently if the cache set failed.
 	 */
 	xcacheSet(params: CacheSetParams): void {
-		return this.notify('xcache/set', params);
+		this.notify('xcache/set', params);
 	}
 
 	/**
@@ -197,7 +197,7 @@ export class RemoteLanguageClient {
 	 *
 	 * @param params The edits to apply.
 	 */
-	workspaceApplyEdit(params: ApplyWorkspaceEditParams, childOf = new Span()): Promise<ApplyWorkspaceEditResponse> {
-		return this.request('workspace/applyEdit', params, childOf).toPromise();
+	workspaceApplyEdit(params: ApplyWorkspaceEditParams, childOf = new Span()): Observable<ApplyWorkspaceEditResponse> {
+		return this.request('workspace/applyEdit', params, childOf);
 	}
 }

src/packages.ts

@@ -8,7 +8,7 @@ import { Disposable } from './disposable';
 import { FileSystemUpdater } from './fs';
 import { Logger, NoopLogger } from './logging';
 import { InMemoryFileSystem } from './memfs';
-import { tracePromise } from './tracing';
+import { traceObservable } from './tracing';
 
 /**
  * Schema of a package.json file
@@ -145,37 +145,44 @@ export class PackageManager extends EventEmitter implements Disposable {
 
 	/**
 	 * Gets the content of the closest package.json known to to the DependencyManager in the ancestors of a URI
+	 *
+	 * @return Observable that emits a single PackageJson or never
 	 */
-	async getClosestPackageJson(uri: string, span = new Span()): Promise<PackageJson | undefined> {
-		await this.updater.ensureStructure();
-		const packageJsonUri = this.getClosestPackageJsonUri(uri);
-		if (!packageJsonUri) {
-			return undefined;
-		}
-		return await this.getPackageJson(packageJsonUri, span);
+	getClosestPackageJson(uri: string, span = new Span()): Observable<PackageJson> {
+		return this.updater.ensureStructure()
+			.concat(Observable.defer(() => {
+				const packageJsonUri = this.getClosestPackageJsonUri(uri);
+				if (!packageJsonUri) {
+					return Observable.empty<never>();
+				}
+				return this.getPackageJson(packageJsonUri, span);
+			}));
 	}
 
 	/**
 	 * Returns the parsed package.json of the passed URI
 	 *
 	 * @param uri URI of the package.json
+	 * @return Observable that emits a single PackageJson or never
 	 */
-	async getPackageJson(uri: string, childOf = new Span()): Promise<PackageJson> {
-		return tracePromise('Get package.json', childOf, async span => {
+	getPackageJson(uri: string, childOf = new Span()): Observable<PackageJson> {
+		return traceObservable('Get package.json', childOf, span => {
 			span.addTags({ uri });
 			if (uri.includes('/node_modules/')) {
-				throw new Error(`Not an own package.json: ${uri}`);
+				return Observable.throw(new Error(`Not an own package.json: ${uri}`));
 			}
 			let packageJson = this.packages.get(uri);
 			if (packageJson) {
-				return packageJson;
+				return Observable.of(packageJson);
 			}
-			await this.updater.ensure(uri, span);
-			packageJson = this.packages.get(uri)!;
-			if (!packageJson) {
-				throw new Error(`Expected ${uri} to be registered in PackageManager`);
-			}
-			return packageJson;
+			return this.updater.ensure(uri, span)
+				.concat(Observable.defer(() => {
+					packageJson = this.packages.get(uri)!;
+					if (!packageJson) {
+						return Observable.throw(new Error(`Expected ${uri} to be registered in PackageManager`));
+					}
+					return Observable.of(packageJson);
+				}));
 		});
 	}