remove pipeline() chain

it cannot be chained afaict

src/backupformat/rsync.js

@@ -13,11 +13,13 @@ const assert = require('assert'),
     async = require('async'),
     BoxError = require('../boxerror.js'),
     DataLayout = require('../datalayout.js'),
+    { DecryptStream } = require('../hush.js'),
     debug = require('debug')('box:backupformat/rsync'),
     fs = require('fs'),
     hush = require('../hush.js'),
     once = require('../once.js'),
     path = require('path'),
+    ProgressStream = require('../progress-stream.js'),
     promiseRetry = require('../promise-retry.js'),
     safe = require('safetydance'),
     storage = require('../storage.js'),
@@ -182,17 +184,27 @@ function downloadDir(backupConfig, backupFilePath, dataLayout, progressCallback,
                 throw downloadError;
             }
 
-            const destStream = hush.createWriteStream(destFilePath, backupConfig.encryption);
+            const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds
-
+            ps.on('progress', function (progress) {
-            destStream.on('progress', function (progress) {
                 const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024);
                 if (!transferred && !speed) return progressCallback({ message: `Downloading ${entry.fullPath}` }); // 0M@0MBps looks wrong
                 progressCallback({ message: `Downloading ${entry.fullPath}: ${transferred}M@${speed}MBps` });
             });
 
+            const destStream = fs.createWriteStream(destFilePath);
+
+            const streams = [ sourceStream, ps ];
+
+            if (backupConfig.encryption) {
+                const decryptStream = new DecryptStream(backupConfig.encryption);
+                streams.push(decryptStream);
+            }
+
+            streams.push(destStream);
+
             progressCallback({ message: `Downloading ${entry.fullPath} to ${destFilePath}` });
 
-            const [pipelineError] = await safe(stream.promises.pipeline(sourceStream, destStream));
+            const [pipelineError] = await safe(stream.promises.pipeline(streams));
             if (pipelineError) {
                 progressCallback({ message: `Download error ${entry.fullPath} to ${destFilePath}: ${pipelineError.message}` });
                 throw pipelineError;

src/hush.js

@@ -6,7 +6,6 @@ const assert = require('assert'),
     debug = require('debug')('box:hush'),
     fs = require('fs'),
     ProgressStream = require('./progress-stream.js'),
-    stream = require('stream'),
     TransformStream = require('stream').Transform;
 
 class EncryptStream extends TransformStream {
@@ -181,35 +180,6 @@ function createReadStream(sourceFile, encryption) {
     }
 }
 
-function createWriteStream(destFile, encryption) {
-    assert.strictEqual(typeof destFile, 'string');
-    assert.strictEqual(typeof encryption, 'object');
-
-    const destStream = fs.createWriteStream(destFile);
-    const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds
-
-    const streams = [ ps ];
-
-    if (encryption) {
-        const decryptStream = new DecryptStream(encryption);
-        streams.push(decryptStream);
-    }
-
-    streams.push(destStream);
-
-    stream.pipeline(streams, function (error) {
-        if (error) {
-            debug(`createWriteStream: write stream error ${destFile}. %o`, error);
-            ps.emit('error', new BoxError(BoxError.FS_ERROR, `Write error ${destFile}: ${error.message}`));
-        } else {
-            debug(`createWriteStream: ${destFile} done`);
-            return ps.emit('done');
-        }
-    });
-
-    return ps;
-}
-
 exports = module.exports = {
     EncryptStream,
     DecryptStream,
@@ -218,5 +188,4 @@ exports = module.exports = {
     decryptFilePath,
 
     createReadStream,
-    createWriteStream
 };