fix(fs/local#create{Read,Write}Stream): wait for file to be opened

This also fix `createOutputStream` when directories are missing.

@xen-orchestra/fs/src/abstract.js

@@ -197,6 +197,21 @@ export default class RemoteHandlerAbstract {
     )
   }
 
+  createWriteStream(
+    file: File,
+    options: { end?: number, flags?: string, start?: number } = {}
+  ): Promise<LaxWritable> {
+    return timeout.call(
+      this._createWriteStream(
+        typeof file === 'string' ? normalizePath(file) : file,
+        {
+          flags: 'wx',
+          ...options,
+        }
+      )
+    )
+  }
+
   // Free the resources possibly dedicated to put the remote at work, when it
   // is no more needed
   async forget(): Promise<void> {

@xen-orchestra/fs/src/fs.spec.js

@@ -84,10 +84,10 @@ handlers.forEach(url => {
     })
 
     describe('#createOutputStream()', () => {
-      testWithFileDescriptor('file', 'wx', async ({ file, flags }) => {
-        const stream = await handler.createOutputStream(file, { flags })
+      it('creates parent dir if missing', async () => {
+        const stream = await handler.createOutputStream('dir/file')
         await fromCallback(cb => pipeline(createTestDataStream(), stream, cb))
-        await expect(await handler.readFile('file')).toEqual(TEST_DATA)
+        await expect(await handler.readFile('dir/file')).toEqual(TEST_DATA)
       })
     })
 
@@ -103,6 +103,19 @@ handlers.forEach(url => {
       })
     })
 
+    describe('#createWriteStream()', () => {
+      testWithFileDescriptor('file', 'wx', async ({ file, flags }) => {
+        const stream = await handler.createWriteStream(file, { flags })
+        await fromCallback(cb => pipeline(createTestDataStream(), stream, cb))
+        await expect(await handler.readFile('file')).toEqual(TEST_DATA)
+      })
+
+      it('fails if parent dir is missing', async () => {
+        const error = await rejectionOf(handler.createWriteStream('dir/file'))
+        expect(error.code).toBe('ENOENT')
+      })
+    })
+
     describe('#getSize()', () => {
       beforeEach(() => handler.outputFile('file', TEST_DATA))
 

@xen-orchestra/fs/src/local.js

@@ -1,4 +1,5 @@
 import fs from 'fs-extra'
+import { fromEvent } from 'promise-toolbox'
 
 import RemoteHandlerAbstract from './abstract'
 
@@ -20,23 +21,29 @@ export default class LocalHandler extends RemoteHandlerAbstract {
   }
 
   async _createReadStream(file, options) {
-    return typeof file === 'string'
-      ? fs.createReadStream(this._getFilePath(file), options)
-      : fs.createReadStream('', {
-          autoClose: false,
-          ...options,
-          fd: file.fd,
-        })
+    if (typeof file === 'string') {
+      const stream = fs.createReadStream(this._getFilePath(file), options)
+      await fromEvent(stream, 'open')
+      return stream
+    }
+    return fs.createReadStream('', {
+      autoClose: false,
+      ...options,
+      fd: file.fd,
+    })
   }
 
   async _createWriteStream(file, options) {
-    return typeof file === 'string'
-      ? fs.createWriteStream(this._getFilePath(file), options)
-      : fs.createWriteStream('', {
-          autoClose: false,
-          ...options,
-          fd: file.fd,
-        })
+    if (typeof file === 'string') {
+      const stream = fs.createWriteStream(this._getFilePath(file), options)
+      await fromEvent(stream, 'open')
+      return stream
+    }
+    return fs.createWriteStream('', {
+      autoClose: false,
+      ...options,
+      fd: file.fd,
+    })
   }
 
   async _getSize(file) {