cancel support for extension install

src/__tests__/modules/extensionDependency.test.ts

@@ -254,11 +254,7 @@ describe('DependenciesInstaller', () => {
     let directory = path.join(os.tmpdir(), uuid())
     dirs.push(directory)
     writeJson(path.join(directory, 'package.json'), { dependencies: { foo: '>= 0.0.1' } })
-    let other = path.join(os.tmpdir(), uuid())
-    dirs.push(other)
-    writeJson(path.join(other, 'package.json'), { dependencies: { bar: '>= 0.0.1' } })
     let one = session.createInstaller(directory, () => {})
-    let two = session.createInstaller(other, () => {})
     let spy = jest.spyOn(one, 'fetchInfos').mockImplementation(() => {
       return new Promise((resolve, reject) => {
         one.token.onCancellationRequested(() => {
@@ -271,20 +267,26 @@ describe('DependenciesInstaller', () => {
       })
     })
     let p = one.installDependencies()
-    let err
-    two.installDependencies().catch(error => {
-      err = error
-    })
     await helper.wait(30)
-    session.cancel()
+    one.cancel()
     let fn = async () => {
       await p
     }
     await expect(fn()).rejects.toThrow(Error)
     spy.mockRestore()
-    await helper.waitValue(() => {
-      return err != null
-    }, true)
+  })
+
+  it('should throw when Cancellation requested', async () => {
+    let install = create(undefined, '')
+    install.cancel()
+    let fn = async () => {
+      await install.fetch(new URL('/', url), { timeout: 10 }, 3)
+    }
+    await expect(fn()).rejects.toThrow(CancellationError)
+    fn = async () => {
+      await install.download(new URL('/', url), 'filename', '')
+    }
+    await expect(fn()).rejects.toThrow(CancellationError)
   })
 
   it('should retry fetch', async () => {

src/extension/dependency.ts

@@ -352,6 +352,7 @@ export class DependenciesInstaller {
   }
 
   public async fetch(url: string | URL, options: FetchOptions, retry = 1): Promise<any> {
+    if (this.tokenSource.token.isCancellationRequested) throw new CancellationError()
     for (let i = 0; i < retry; i++) {
       try {
         return await fetch(url, options, this.tokenSource.token)
@@ -369,6 +370,7 @@ export class DependenciesInstaller {
    * Download tgz file with sha1 check.
    */
   public async download(url: string | URL, filename: string, shasum: string, retry = 1, timeout?: number): Promise<string> {
+    if (this.tokenSource.token.isCancellationRequested) throw new CancellationError()
     for (let i = 0; i < retry; i++) {
       try {
         let fullpath = path.join(this.dest, filename)
@@ -395,32 +397,18 @@ export class DependenciesInstaller {
 
   public cancel(): void {
     this.tokenSource.cancel()
-    this.tokenSource = new CancellationTokenSource()
   }
 }
 
 export class DependencySession {
   private resolvedInfos: Map<string, ModuleInfo> = new Map()
-  private installers: Set<DependenciesInstaller> = new Set()
   constructor(
     public readonly registry: URL,
     public readonly modulesRoot: string
   ) {
   }
 
   public createInstaller(directory: string, onMessage: (msg: string) => void): DependenciesInstaller {
-    let installer = new DependenciesInstaller(this.registry, this.resolvedInfos, this.modulesRoot, directory, onMessage)
-    this.installers.add(installer)
-    return installer
-  }
-
-  /**
-   * Cancel all installer
-   */
-  public cancel(): void {
-    for (let item of this.installers) {
-      item.cancel()
-    }
-    this.installers.clear()
+    return new DependenciesInstaller(this.registry, this.resolvedInfos, this.modulesRoot, directory, onMessage)
   }
 }

src/extension/index.ts

@@ -168,8 +168,16 @@ export class Extensions {
     let disposables: Disposable[] = this.disposables = []
     let installBuffer = this.createInstallerUI(false, false, disposables)
     let tokenSource = new CancellationTokenSource()
+    let installers: Map<string, IInstaller> = new Map()
+    installBuffer.onDidCancel(key => {
+      let item = installers.get(key)
+      if (item) item.dispose()
+    })
     disposables.push(Disposable.create(() => {
       tokenSource.cancel()
+      for (let item of installers.values()) {
+        item.dispose()
+      }
     }))
     await Promise.resolve(installBuffer.start(list))
     let registry = await registryUrl()
@@ -178,28 +186,24 @@ export class Extensions {
       try {
         installBuffer.startProgress(key)
         installer = this.createInstaller(registry, key)
-        disposables.push(installer)
+        installers.set(key, installer)
         installer.on('message', (msg, isProgress) => {
           installBuffer.addMessage(key, msg, isProgress)
         })
+        logger.debug('install:', key)
         let result = await installer.install()
-        disposables = disposables.filter(o => o !== installer)
         installBuffer.finishProgress(key, true)
         this.states.addExtension(result.name, result.url ? result.url : `>=${result.version}`)
         let ms = key.match(/@[\d.]+$/)
         if (ms != null) this.states.setLocked(result.name, true)
         await this.manager.loadExtension(result.folder)
       } catch (err: any) {
-        installBuffer.addMessage(key, err.message)
-        installBuffer.finishProgress(key, false)
-        if (!isCancellationError(err)) {
-          void window.showErrorMessage(`Error on install ${key}: ${err}`)
-          logger.error(`Error on install ${key}`, err)
-        }
+        this.onInstallError(key, installBuffer, err)
       }
     }
     await concurrent(list, fn, 3, tokenSource.token)
-    disposables.splice(0, disposables.length)
+    let len = disposables.length
+    disposables.splice(0, len)
   }
 
   /**
@@ -219,11 +223,19 @@ export class Extensions {
     this.cleanModulesFolder()
     let registry = await registryUrl()
     let disposables: Disposable[] = this.disposables = []
+    let installers: Map<string, IInstaller> = new Map()
     let installBuffer = this.createInstallerUI(true, silent, disposables)
     let tokenSource = new CancellationTokenSource()
     disposables.push(Disposable.create(() => {
       tokenSource.cancel()
+      for (let item of installers.values()) {
+        item.dispose()
+      }
     }))
+    installBuffer.onDidCancel(key => {
+      let item = installers.get(key)
+      if (item) item.dispose()
+    })
     await Promise.resolve(installBuffer.start(stats.map(o => o.id)))
     let fn = async (stat: ExtensionInfo): Promise<void> => {
       let { id } = stat
@@ -232,27 +244,30 @@ export class Extensions {
         installBuffer.startProgress(id)
         let url = stat.exotic ? stat.uri : null
         installer = this.createInstaller(registry, id)
-        disposables.push(installer)
+        installers.set(id, installer)
         installer.on('message', (msg, isProgress) => {
           installBuffer.addMessage(id, msg, isProgress)
         })
         let directory = await installer.update(url)
-        disposables = disposables.filter(o => o !== installer)
         installBuffer.finishProgress(id, true)
         if (directory) await this.manager.loadExtension(directory)
       } catch (err: any) {
-        installBuffer.addMessage(id, err.message)
-        installBuffer.finishProgress(id, false)
-        if (!isCancellationError(err)) {
-          void window.showErrorMessage(`Error on update ${id}: ${err}`)
-          logger.error(`Error on update ${id}`, err)
-        }
+        this.onInstallError(id, installBuffer, err)
       }
     }
     await concurrent(stats, fn, silent ? 1 : 3, tokenSource.token)
     disposables.splice(0, disposables.length)
   }
 
+  private onInstallError(id: string, installBuffer: InstallUI, err: Error): void {
+    installBuffer.addMessage(id, err.message)
+    installBuffer.finishProgress(id, false)
+    if (!isCancellationError(err)) {
+      void window.showErrorMessage(`Error on install ${id}: ${err}`)
+      logger.error(`Error on update ${id}`, err)
+    }
+  }
+
   /**
    * Get all extension states
    */

src/extension/installer.ts

@@ -165,6 +165,7 @@ export class Installer extends EventEmitter implements IInstaller {
   }
 
   public async install(): Promise<InstallResult> {
+    this.emit('message', `fetch info of ${this.def}`, false)
     let info = await this.getInfo()
     logger.info(`Fetched info of ${this.def}`, info)
     let { name, version } = info
@@ -254,6 +255,5 @@ export class Installer extends EventEmitter implements IInstaller {
 
   public dispose(): void {
     this.tokenSource.cancel()
-    this.tokenSource = new CancellationTokenSource()
   }
 }

src/extension/ui.ts

@@ -1,6 +1,6 @@
 'use strict'
 import { debounce } from 'debounce'
-import { Disposable } from 'vscode-languageserver-protocol'
+import { Disposable, Emitter, Event } from 'vscode-languageserver-protocol'
 import events from '../events'
 import { frames } from '../model/status'
 import { HighlightItem, OutputChannel } from '../types'
@@ -18,13 +18,16 @@ export enum State {
 }
 
 export interface InstallUI {
+  onDidCancel: Event<string>
   start(names: string[]): void | Promise<void>
   addMessage(name: string, msg: string, isProgress?: boolean): void
   startProgress(name: string): void
   finishProgress(name: string, succeed?: boolean): void
 }
 
 export class InstallChannel implements InstallUI {
+  private readonly _onDidCancel = new Emitter<string>()
+  public readonly onDidCancel: Event<string> = this._onDidCancel.event
   constructor(private isUpdate: boolean, private channel: OutputChannel) {
   }
 
@@ -59,6 +62,8 @@ export class InstallBuffer implements InstallUI {
   private names: string[] = []
   private interval: NodeJS.Timer
   public bufnr: number
+  private readonly _onDidCancel = new Emitter<string>()
+  public readonly onDidCancel: Event<string> = this._onDidCancel.event
 
   constructor(private isUpdate: boolean, onClose = () => {}) {
     let floatFactory = window.createFloatFactory({ modes: ['n'] })
@@ -163,6 +168,21 @@ export class InstallBuffer implements InstallUI {
     return this.interval == null
   }
 
+  public async onTab(): Promise<void> {
+    let line = await workspace.nvim.eval(`line(".")-1`) as number
+    let name = this.names[line - 2]
+    if (!name) return
+    let state = this.statMap.get(name)
+    let couldCancel = state === State.Progressing
+    let actions: string[] = []
+    if (couldCancel) actions.push('Cancel')
+    if (actions.length === 0) return
+    let idx = await window.showMenuPicker(['Cancel'])
+    if (idx === 0) {
+      this._onDidCancel.fire(name)
+    }
+  }
+
   // draw frame
   public draw(): void {
     let { remains, bufnr } = this
@@ -198,6 +218,7 @@ export class InstallBuffer implements InstallUI {
     if (!isSync) nvim.command('nnoremap <silent><nowait><buffer> q :q<CR>', true)
     this.highlight()
     let res = await nvim.resumeNotification()
+    workspace.registerLocalKeymap('n', '<tab>', this.onTab.bind(this))
     this.bufnr = res[0][1] as number
     this.interval = setInterval(() => {
       this.draw()

src/util/index.ts

@@ -160,28 +160,28 @@ export function delay(func: () => void, defaultDelay: number): ((ms?: number) =>
   return fn as any
 }
 
-export function concurrent<T>(arr: T[], fn: (val: T) => Promise<void>, limit = 3, token?: CancellationToken): Promise<void> {
+export function concurrent<T>(arr: ReadonlyArray<T>, fn: (val: T) => Promise<void>, limit = 3, token?: CancellationToken): Promise<void> {
   if (arr.length == 0) return Promise.resolve()
   let finished = 0
   let total = arr.length
-  let remain = arr.slice()
+  let curr = 0
   return new Promise(resolve => {
     let run = (val): void => {
       if (token && token.isCancellationRequested) return resolve()
       let cb = () => {
         finished = finished + 1
         if (finished == total) {
           resolve()
-        } else if (remain.length) {
-          let next = remain.shift()
-          run(next)
+        } else if (curr < total - 1) {
+          curr++
+          run(arr[curr])
         }
       }
       fn(val).then(cb, cb)
     }
-    for (let i = 0; i < Math.min(limit, remain.length); i++) {
-      let val = remain.shift()
-      run(val)
+    curr = Math.min(limit, total) - 1
+    for (let i = 0; i <= curr; i++) {
+      run(arr[i])
     }
   })
 }