feat(fs/outputStream): validator support

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

@@ -204,8 +204,9 @@ export default class RemoteHandlerAbstract {
    * @param {object} [options]
    * @param {boolean} [options.checksum]
    * @param {number} [options.dirMode]
+   * @param {(this: RemoteHandlerAbstract, path: string) => Promise<undefined>} [options.validator] Function that will be called before the data is commited to the remote, if it fails, file should not exist
    */
-  async outputStream(path, input, { checksum = true, dirMode } = {}) {
+  async outputStream(path, input, { checksum = true, dirMode, validator } = {}) {
     path = normalizePath(path)
     let checksumStream
     if (checksum) {
@@ -215,6 +216,7 @@ export default class RemoteHandlerAbstract {
     }
     await this._outputStream(path, await input, {
       dirMode,
+      validator,
     })
     if (checksum) {
       await this._outputFile(checksumFile(path), await checksumStream.checksum, { dirMode, flags: 'wx' })
@@ -482,13 +484,16 @@ export default class RemoteHandlerAbstract {
     return this._outputFile(file, data, { flags })
   }
 
-  async _outputStream(path, input, { dirMode }) {
+  async _outputStream(path, input, { dirMode, validator }) {
     const tmpPath = `${dirname(path)}/.${basename(path)}`
     const output = await this.createOutputStream(tmpPath, {
       dirMode,
     })
     try {
       await fromCallback(pipeline, input, output)
+      if (validator !== undefined) {
+        await validator.call(this, path)
+      }
       await this.rename(tmpPath, path)
     } catch (error) {
       await this.unlink(tmpPath)

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

@@ -51,7 +51,7 @@ export default class S3Handler extends RemoteHandlerAbstract {
     return { Bucket: this._bucket, Key: this._dir + file }
   }
 
-  async _outputStream(path, input) {
+  async _outputStream(path, input, { validator }) {
     await this._s3.upload(
       {
         ...this._createParams(path),
@@ -59,6 +59,14 @@ export default class S3Handler extends RemoteHandlerAbstract {
       },
       { partSize: IDEAL_FRAGMENT_SIZE, queueSize: 1 }
     )
+    if (validator !== undefined) {
+      try {
+        await validator.call(this, path)
+      } catch (error) {
+        await this.unlink(path)
+        throw error
+      }
+    }
   }
 
   async _writeFile(file, data, options) {

CHANGELOG.unreleased.md

@@ -28,6 +28,6 @@
 >
 > In case of conflict, the highest (lowest in previous list) `$version` wins.
 
-- @xen-orchestra/fs patch
+- @xen-orchestra/fs minor
 - @xen-orchestra/backups patch
 - xo-server patch