WiP

packages/xen-api/examples/import-vdi

@@ -39,7 +39,12 @@ defer(async ($defer, argv) => {
   $defer(() => xapi.disconnect())
 
   const { cancel, token } = CancelToken.source()
-  process.on('SIGINT', cancel)
+  process.once('SIGINT', () => {
+    cancel()
+    process.once('SIGINT', () => {
+      process.exit(1)
+    })
+  })
 
   let input = createInputStream(opts._[1])
   $defer.onFailure(() => input.destroy())
@@ -75,7 +80,5 @@ defer(async ($defer, argv) => {
     },
   })
 
-  if (result !== undefined) {
-    console.log(result)
-  }
+  console.log(result !== undefined ? result : 'ok')
 })(process.argv.slice(2)).catch(console.error.bind(console, 'error'))

packages/xen-api/src/index.js

@@ -11,8 +11,18 @@ import { coalesceCalls } from '@vates/coalesce-calls'
 import { Collection } from 'xo-collection'
 import { EventEmitter } from 'events'
 import { Index } from 'xo-collection/index'
-import { cancelable, defer, fromCallback, ignoreErrors, pDelay, pRetry, pTimeout } from 'promise-toolbox'
 import { limitConcurrency } from 'limit-concurrency-decorator'
+import {
+  cancelable,
+  CancelToken,
+  defer,
+  fromCallback,
+  fromEvent,
+  ignoreErrors,
+  pDelay,
+  pRetry,
+  pTimeout,
+} from 'promise-toolbox'
 
 import autoTransport from './transports/auto'
 import debug from './_debug'
@@ -90,6 +100,7 @@ export class Xapi extends EventEmitter {
       opts.syncStackTraces ?? process.env.NODE_ENV === 'development' ? addSyncStackTrace : identity
     this._callTimeout = makeCallSetting(opts.callTimeout, 60 * 60 * 1e3) // 1 hour but will be reduced in the future
     this._httpInactivityTimeout = opts.httpInactivityTimeout ?? 5 * 60 * 1e3 // 5 mins
+    this._httpTimeout = opts.httpTimeout ?? 24 * 60 * 60 * 1e3 // 24 hours
     this._eventPollDelay = opts.eventPollDelay ?? 60 * 1e3 // 1 min
     this._pool = null
     this._readOnly = Boolean(opts.readOnly)
@@ -369,6 +380,13 @@ export class Xapi extends EventEmitter {
 
   @cancelable
   async getResource($cancelToken, pathname, { host, query, task } = {}) {
+    const timeout = this._httpTimeout
+    if (timeout !== 0) {
+      const source = CancelToken.source([$cancelToken])
+      setTimeout(source.cancel, timeout)
+      $cancelToken = source.token
+    }
+
     const taskRef = await this._autoTask(task, `Xapi#getResource ${pathname}`)
 
     query = { ...query, session_id: this.sessionId }
@@ -431,6 +449,13 @@ export class Xapi extends EventEmitter {
       throw new Error('cannot put resource in read only mode')
     }
 
+    const timeout = this._httpTimeout
+    if (timeout !== 0) {
+      const source = CancelToken.source([$cancelToken])
+      setTimeout(source.cancel, timeout)
+      $cancelToken = source.token
+    }
+
     const taskRef = await this._autoTask(task, `Xapi#putResource ${pathname}`)
 
     query = { ...query, session_id: this.sessionId }
@@ -519,6 +544,10 @@ export class Xapi extends EventEmitter {
           )
         : doRequest(url.href)
     )
+    const responseEnd = fromEvent(response, 'end')
+    responseEnd.catch(noop)
+
+    console.log({ useHack })
 
     if (pTaskResult !== undefined) {
       if (useHack) {
@@ -539,10 +568,12 @@ export class Xapi extends EventEmitter {
 
     const { req } = response
     if (!req.finished) {
+      console.log('waiting for request to finish')
       await new Promise((resolve, reject) => {
         req.on('finish', resolve).on('error', reject)
         response.on('error', reject)
       })
+      console.log('request finished')
     }
 
     if (useHack) {