feat: watch for file changes

closes: Old versions of files appear #38

src/parquet-document-provider.ts

@@ -1,23 +1,33 @@
 import * as vscode from "vscode";
-import * as assert from 'assert';
+import ParquetDocument from './parquet-document';
 
 export class ParquetTextDocumentContentProvider implements vscode.TextDocumentContentProvider {
-  private static jsonMap: Map<string, string> = new Map();
+  private _onDidChange = new vscode.EventEmitter<vscode.Uri>();
+  private _documents = new Map<vscode.Uri, ParquetDocument>();
+  private _subscriptions: vscode.Disposable;
 
-  public static has(path: string): boolean {
-    return ParquetTextDocumentContentProvider.jsonMap.has(path);
+  constructor() {
+    this._subscriptions = vscode.workspace.onDidCloseTextDocument(doc => this._documents.delete(doc.uri));
   }
 
-  public static add(path:string, content: string): void {
-    ParquetTextDocumentContentProvider.jsonMap.set(path, content);
+  dispose() {
+    this._subscriptions.dispose();
+    this._documents.clear();
+    this._onDidChange.dispose();
   }
 
-  onDidChangeEmitter = new vscode.EventEmitter<vscode.Uri>();
-  onDidChange = this.onDidChangeEmitter.event;
+  get onDidChange() {
+    return this._onDidChange.event;
+  }
 
   async provideTextDocumentContent(uri: vscode.Uri): Promise<string | undefined> {
-    const parquetPath = uri.fsPath.replace(/\.as\.json$/, '');
-    assert(ParquetTextDocumentContentProvider.has(parquetPath));
-    return ParquetTextDocumentContentProvider.jsonMap.get(parquetPath);
+    // already loaded?
+    const document = this._documents.get(uri) || (await (async _ => {
+      const document = await ParquetDocument.create(uri, this._onDidChange);
+      this._documents.set(uri, document);
+      return document;
+    })());
+
+    return document.value;
   }
 }

src/parquet-document.ts

@@ -0,0 +1,71 @@
+import * as os from 'os';
+import * as path from 'path';
+import * as vscode from 'vscode';
+import { ParquetBackend } from "./parquet-backend";
+import { ParquetToolsBackend } from './parquet-tools-backend';
+import { ParquetsBackend } from './parquets-backend';
+import { useParquetTools } from "./settings";
+
+export default class ParquetDocument implements vscode.Disposable {
+  private readonly _uri: vscode.Uri;
+  private readonly _emitter: vscode.EventEmitter<vscode.Uri>;
+
+  private _lines: string[] = [];
+  private readonly _disposables: vscode.Disposable[] = [];
+  private readonly _parquetPath: string;
+  private readonly _backend: ParquetBackend = useParquetTools() ? new ParquetToolsBackend() : new ParquetsBackend();
+
+
+  private constructor(uri: vscode.Uri, emitter: vscode.EventEmitter<vscode.Uri>) {
+    this._uri = uri;
+    this._emitter = emitter;
+    this._parquetPath = this._uri.fsPath.replace(/\.as\.json$/, '');
+    const watcher = vscode.workspace.createFileSystemWatcher(new vscode.RelativePattern(this._parquetPath, "*"));
+    this._disposables.push(watcher);
+    this._disposables.push(watcher.onDidChange(async _ => {
+      return await this._populate();
+    }));
+    this._disposables.push(watcher.onDidCreate(async _ => {
+      return await this._populate();
+    }));
+  }
+
+  dispose() {
+    for (const disposable of this._disposables) {
+      disposable.dispose();
+    }
+  }
+
+  public static async create(uri: vscode.Uri, emitter: vscode.EventEmitter<vscode.Uri>): Promise<ParquetDocument> {
+    const me = new ParquetDocument(uri, emitter);
+    await me._populate();
+    return me;
+  }
+
+  get value() {
+    return this._lines.join(os.EOL) + os.EOL;
+  }
+
+  private async _populate() {
+    const lines: string[] = [];
+
+    try {
+      await vscode.window.withProgress({
+        location: vscode.ProgressLocation.Notification,
+        title: `opening ${path.basename(this._parquetPath)}`,
+        cancellable: true
+      },
+        async (progress, token) => {
+          for await (const line of this._backend.toJson(this._parquetPath, token)) {
+            lines.push(line);
+          }
+        });
+    } catch (err) {
+      await vscode.window.showErrorMessage(`${err}`);
+    }
+    if (lines != this._lines) {
+      this._lines = lines;
+      this._emitter.fire(this._uri);
+    }
+  }
+}

src/parquet-editor-provider.ts

@@ -3,66 +3,26 @@ import * as path from 'path';
 import { getNonce } from './util';
 import { Disposable } from "./dispose";
 import { ParquetTextDocumentContentProvider } from './parquet-document-provider';
-import { ParquetToolsBackend } from './parquet-tools-backend';
-import { ParquetsBackend } from './parquets-backend';
-import toArray from '@async-generators/to-array';
 import { getLogger } from './logger';
-import { useParquetTools } from "./settings";
-import { ParquetBackend } from "./parquet-backend";
-
-class ParquetDocument extends Disposable implements vscode.CustomDocument {
+class CustomParquetDocument extends Disposable implements vscode.CustomDocument {
   uri: vscode.Uri;
   path: string;
-  backend: ParquetBackend;
 
   constructor(uri: vscode.Uri) {
     super();
     this.uri = uri;
     this.path = uri.fsPath;
-    this.backend = useParquetTools() ? new ParquetToolsBackend() : new ParquetsBackend();
   }
 
-  private async open() {
+  public async open() {
     getLogger().info(`opening ${this.path}.as.json`);
     await vscode.window.showTextDocument(
       this.uri.with({ scheme: 'parquet', path: this.path + '.as.json' })
     );
   }
-
-  private async * toJson(parquetPath: string, token?: vscode.CancellationToken): AsyncGenerator<string, void, undefined> {
-    yield* this.backend.toJson(parquetPath, token);
-  }
-
-  public async show() {
-    getLogger().info(`showing ${this.path}.as.json`);
-    if (ParquetTextDocumentContentProvider.has(this.path)) {
-      return await this.open();
-    }
-
-    try {
-      return await vscode.window.withProgress({
-        location: vscode.ProgressLocation.Notification,
-        title: `opening ${path.basename(this.path)}`,
-        cancellable: true
-      },
-        async (progress, token) => {
-          const json = await toArray(this.toJson(this.path, token));
-          if (!token.isCancellationRequested) {
-            ParquetTextDocumentContentProvider.add(this.path, json.join(''));
-            await this.open();
-          }
-        });
-    } catch (err) {
-      await vscode.window.showErrorMessage(`${err}`);
-    }
-  }
-
-  dispose(): void {
-    super.dispose();
-  }
 }
 
-export class ParquetEditorProvider implements vscode.CustomReadonlyEditorProvider<ParquetDocument> {
+export class ParquetEditorProvider implements vscode.CustomReadonlyEditorProvider<CustomParquetDocument> {
 
   private static readonly viewType = 'parquetViewer.parquetViewer';
 
@@ -78,12 +38,12 @@ export class ParquetEditorProvider implements vscode.CustomReadonlyEditorProvide
     return providerRegistration;
   }
 
-  async openCustomDocument(uri: vscode.Uri): Promise<ParquetDocument> {
-    return new ParquetDocument(uri);
+  async openCustomDocument(uri: vscode.Uri): Promise<CustomParquetDocument> {
+    return new CustomParquetDocument(uri);
   }
 
   async resolveCustomEditor(
-    document: ParquetDocument,
+    document: CustomParquetDocument,
     webviewPanel: vscode.WebviewPanel,
     _token: vscode.CancellationToken
   ): Promise<void> {
@@ -96,18 +56,18 @@ export class ParquetEditorProvider implements vscode.CustomReadonlyEditorProvide
 
     webviewPanel.webview.onDidReceiveMessage(e => this.onMessage(document, e));
 
-    await document.show();
+    await document.open();
   }
 
-  private async onMessage(document: ParquetDocument, message: string) {
+  private async onMessage(document: CustomParquetDocument, message: string) {
     switch (message) {
       case 'clicked':
-        await document.show();
+        await document.open();
         break;
     }
   }
 
-  private getHtmlForWebview(webview: vscode.Webview, document: ParquetDocument): string {
+  private getHtmlForWebview(webview: vscode.Webview, document: CustomParquetDocument): string {
     // Use a nonce to whitelist which scripts can be run
     const nonce = getNonce();
 

src/parquet-tools-backend.ts

@@ -5,12 +5,11 @@ import * as assert from 'assert';
 import { getLogger } from './logger';
 import { parquetTools as getParquetTools } from './settings';
 import { createInterface } from 'readline';
-import * as os from 'os';
 import { ParquetBackend } from './parquet-backend';
 
 export class ParquetToolsBackend implements ParquetBackend {
 
-  public static async* spawnParquetTools(params: string[], token?: vscode.CancellationToken): AsyncGenerator<string, string, undefined> {
+  public static async* spawnParquetTools(params: string[], token?: vscode.CancellationToken): AsyncGenerator<string> {
     let parquetTools = getParquetTools();
     if (!parquetTools) {
       throw Error(`illegal value for parquet-viewer.parquetToolsPath setting: ${parquetTools}`);
@@ -36,7 +35,9 @@ export class ParquetToolsBackend implements ParquetBackend {
     });
 
     childProcess.stdout.setEncoding('utf-8');
-    yield* createInterface({input: childProcess.stdout});
+    for await (const line of createInterface({input: childProcess.stdout})) {
+      yield line;
+    }
     let stderr = '';
     childProcess.stderr.on('data', data => stderr += data);
     const code = await new Promise((resolve, reject) => {
@@ -65,9 +66,7 @@ export class ParquetToolsBackend implements ParquetBackend {
     });
 
     try {
-      for await (const line of ParquetToolsBackend.spawnParquetTools(['cat', '-j', parquetPath], token)) {
-        yield `${line}${os.EOL}`
-      }
+      yield* ParquetToolsBackend.spawnParquetTools(['cat', '-j', parquetPath], token);
     } catch (e) {
       let message = `while reading ${parquetPath}: `;
       message += (_ => {

src/parquets-backend.ts

@@ -1,7 +1,6 @@
 import * as vscode from 'vscode';
 import { getLogger } from './logger';
 import { ParquetReader } from '@dvirtz/parquets';
-import * as os from 'os';
 import { ParquetBackend } from './parquet-backend';
 import { jsonSpace } from './settings';
 
@@ -21,7 +20,7 @@ export class ParquetsBackend implements ParquetBackend {
       // read all records from the file and print them
       let record = null;
       while (!token?.isCancellationRequested && (record = await cursor.next())) {
-        yield `${JSON.stringify(record, null, jsonSpace())}${os.EOL}`;
+        yield JSON.stringify(record, null, jsonSpace());
       }
 
       await reader.close();

test/integration/parquet-editor-provider.test.ts

@@ -1,9 +1,9 @@
-import { readFileSync } from 'fs';
 import * as vscode from 'vscode';
 import { ParquetTextDocumentContentProvider } from "../../src/parquet-document-provider";
 import { ParquetEditorProvider } from "../../src/parquet-editor-provider";
-import { getUri } from "./utils";
 import * as settings from '../../src/settings';
+import { getUri, readFile, deferred } from "./utils";
+import * as path from 'path';
 
 jest.mock('../../src/settings', () => {
   const originalModule = jest.requireActual('../../src/settings');
@@ -18,9 +18,15 @@ jest.mock('../../src/settings', () => {
 
 describe('ParquetEditorProvider', function () {
   const disposables: vscode.Disposable[] = [];
+  const workspace = vscode.workspace;
+  const fs = workspace.fs;
+  const editorProvider = new ParquetEditorProvider;
 
   beforeEach(async function () {
     await vscode.commands.executeCommand('workbench.action.closeAllEditors');
+    disposables.push(
+      workspace.registerTextDocumentContentProvider('parquet', new ParquetTextDocumentContentProvider)
+    );
   });
 
   afterEach(function () {
@@ -29,22 +35,54 @@ describe('ParquetEditorProvider', function () {
     }
   });
 
-  test.each([false, true])('shows parquet using parquet tools %p', async function (useParquetTools) {
-    const parquet = await getUri('small.parquet');
-    const json = await getUri('small.json');
-    let eventCalled = false;
-    disposables.push(vscode.window.onDidChangeActiveTextEditor(testEditor => {
-      if (testEditor?.document.fileName === `${parquet.fsPath}.as.json`) {
-        eventCalled = true;
-        expect(testEditor.document.getText()).toEqual(readFileSync(json.fsPath));
-      }
-    }));
-    disposables.push(
-      vscode.workspace.registerTextDocumentContentProvider('parquet', new ParquetTextDocumentContentProvider)
-    );
+  test.each([
+    ['small', true],
+    ['large', false]
+  ])('shows parquet using parquet tools %p', async function (name, useParquetTools) {
+    const parquet = await getUri(`${name}.parquet`);
+    const checkChanged = jest.fn(async function (document: vscode.TextDocument) {
+      expect(document.fileName).toBe(`${parquet.fsPath}.as.json`);
+      const expected = await readFile(`${name}.json`);
+      expect(document.getText()).toEqual(expected);
+    });
+    disposables.push(workspace.onDidOpenTextDocument(checkChanged));
     jest.mocked(settings.useParquetTools).mockReturnValue(useParquetTools);
-    const document = await new ParquetEditorProvider().openCustomDocument(parquet);
-    await document.show();
-    expect(eventCalled).toBeTruthy();
+    const document = await editorProvider.openCustomDocument(parquet);
+    await document.open();
+    expect(checkChanged).toBeCalled();
+  });
+
+  test('updated on file change', async function () {
+    const small = await getUri('small.parquet');
+    const temp = vscode.Uri.file(path.join(path.dirname(small.fsPath), 'temp.parquet'));
+    await fs.copy(small, temp, { overwrite: true });
+
+    const checkChanged = jest.fn(async function (document: vscode.TextDocument, name: string) {
+      expect(document.fileName).toBe(`${temp.fsPath}.as.json`);
+      const expected = await readFile(`${name}.json`);
+      expect(document.getText()).toEqual(expected);
+    });
+
+    const checkSmall = workspace.onDidOpenTextDocument(async document => {
+      await checkChanged(document, 'small');
+    });
+    await (await editorProvider.openCustomDocument(temp)).open();
+    checkSmall.dispose();
+
+    const {promise, resolve} = deferred<void>();
+
+    const checkLarge = workspace.onDidChangeTextDocument(async event => {
+      await checkChanged(event.document, 'large');
+      resolve();
+    });
+    await fs.copy(await getUri('large.parquet'), temp, { overwrite: true });
+
+    await promise;
+
+    checkLarge.dispose();
+
+    expect(checkChanged).toBeCalledTimes(2);
+
+    await workspace.fs.delete(temp);
   });
 });