fix: set the FileResultStreamConverter explicitly

The `SendFilesStream` used to run the `FileResultStreamConverter`
automatically. With making it an option to use a converter,
`SendFilesStream` can be used outside of the Files API, e.g. for
the DAG API.

Closes #696.

src/files/add-pull-stream.js

@@ -1,6 +1,13 @@
 'use strict'
 
 const SendFilesStream = require('../utils/send-files-stream')
+const FileResultStreamConverter = require('../utils/file-result-stream-converter')
 const toPull = require('stream-to-pull-stream')
 
-module.exports = (send) => (options) => toPull(SendFilesStream(send, 'add')(options))
+module.exports = (send) => {
+  return (options) => {
+    options = options || {}
+    options.converter = FileResultStreamConverter
+    return toPull(SendFilesStream(send, 'add')(options))
+  }
+}

src/files/add-readable-stream.js

@@ -1,5 +1,12 @@
 'use strict'
 
 const SendFilesStream = require('../utils/send-files-stream')
+const FileResultStreamConverter = require('../utils/file-result-stream-converter')
 
-module.exports = (send) => SendFilesStream(send, 'add')
+module.exports = (send) => {
+  return (options) => {
+    options = options || {}
+    options.converter = FileResultStreamConverter
+    return SendFilesStream(send, 'add')(options)
+  }
+}

src/files/add.js

@@ -5,6 +5,7 @@ const ConcatStream = require('concat-stream')
 const once = require('once')
 const isStream = require('is-stream')
 const OtherBuffer = require('buffer').Buffer
+const FileResultStreamConverter = require('../utils/file-result-stream-converter')
 const SendFilesStream = require('../utils/send-files-stream')
 
 module.exports = (send) => {
@@ -21,6 +22,7 @@ module.exports = (send) => {
     if (!options) {
       options = {}
     }
+    options.converter = FileResultStreamConverter
 
     const ok = Buffer.isBuffer(_files) ||
                isStream.readable(_files) ||

src/files/write.js

@@ -3,6 +3,7 @@
 const promisify = require('promisify-es6')
 const concatStream = require('concat-stream')
 const once = require('once')
+const FileResultStreamConverter = require('../utils/file-result-stream-converter')
 const SendFilesStream = require('../utils/send-files-stream')
 
 module.exports = (send) => {
@@ -28,7 +29,8 @@ module.exports = (send) => {
 
     const options = {
       args: pathDst,
-      qs: opts
+      qs: opts,
+      converter: FileResultStreamConverter
     }
 
     const stream = sendFilesStream(options)

src/util/fs-add.js

@@ -4,6 +4,7 @@ const isNode = require('detect-node')
 const promisify = require('promisify-es6')
 const moduleConfig = require('../utils/module-config')
 const SendOneFile = require('../utils/send-one-file-multiple-results')
+const FileResultStreamConverter = require('../utils/file-result-stream-converter')
 
 module.exports = (arg) => {
   const sendOneFile = SendOneFile(moduleConfig(arg), 'add')
@@ -31,6 +32,10 @@ module.exports = (arg) => {
       return callback(new Error('"path" must be a string'))
     }
 
-    sendOneFile(path, { qs: opts }, callback)
+    const requestOpts = {
+      qs: opts,
+      converter: FileResultStreamConverter
+    }
+    sendOneFile(path, requestOpts, callback)
   })
 }

src/util/url-add.js

@@ -5,6 +5,7 @@ const parseUrl = require('url').parse
 const request = require('../utils/request')
 const moduleConfig = require('../utils/module-config')
 const SendOneFile = require('../utils/send-one-file-multiple-results')
+const FileResultStreamConverter = require('../utils/file-result-stream-converter')
 
 module.exports = (arg) => {
   const sendOneFile = SendOneFile(moduleConfig(arg), 'add')
@@ -49,7 +50,11 @@ const requestWithRedirect = (url, opts, sendOneFile, callback) => {
       }
       requestWithRedirect(redirection, opts, sendOneFile, callback)
     } else {
-      sendOneFile(res, { qs: opts }, callback)
+      const requestOpts = {
+        qs: opts,
+        converter: FileResultStreamConverter
+      }
+      sendOneFile(res, requestOpts, callback)
     }
   }).end()
 }

src/utils/converter.js → src/utils/file-result-stream-converter.js

@@ -6,7 +6,7 @@ const TransformStream = require('readable-stream').Transform
   Transforms a stream of {Name, Hash} objects to include size
   of the DAG object.
 
-  Usage: inputStream.pipe(new Converter())
+  Usage: inputStream.pipe(new FileResultStreamConverter())
 
   Input object format:
   {
@@ -22,7 +22,7 @@ const TransformStream = require('readable-stream').Transform
     size: 20
   }
 */
-class ConverterStream extends TransformStream {
+class FileResultStreamConverter extends TransformStream {
   constructor (options) {
     const opts = Object.assign({}, options || {}, { objectMode: true })
     super(opts)
@@ -41,4 +41,4 @@ class ConverterStream extends TransformStream {
   }
 }
 
-module.exports = ConverterStream
+module.exports = FileResultStreamConverter

src/utils/send-files-stream.js

@@ -6,7 +6,6 @@ const isStream = require('is-stream')
 const once = require('once')
 const prepareFile = require('./prepare-file')
 const Multipart = require('./multipart')
-const Converter = require('./converter')
 
 function headers (file) {
   const name = file.path
@@ -127,15 +126,27 @@ module.exports = (send, path) => {
 
       response.on('error', (err) => retStream.emit('error', err))
 
-      response.on('data', (d) => {
-        if (d.Bytes && options.progress) {
-          options.progress(d.Bytes)
-        }
-      })
-      const convertedResponse = new Converter()
-      convertedResponse.once('end', () => retStream.push(null))
-      convertedResponse.on('data', (d) => retStream.push(d))
-      response.pipe(convertedResponse)
+      if (options.converter) {
+        response.on('data', (d) => {
+          if (d.Bytes && options.progress) {
+            options.progress(d.Bytes)
+          }
+        })
+
+        const Converter = options.converter
+        const convertedResponse = new Converter()
+        convertedResponse.once('end', () => retStream.push(null))
+        convertedResponse.on('data', (d) => retStream.push(d))
+        response.pipe(convertedResponse)
+      } else {
+        response.on('data', (d) => {
+          if (d.Bytes && options.progress) {
+            options.progress(d.Bytes)
+          }
+          retStream.push(d)
+        })
+        response.once('end', () => retStream.push(null))
+      }
     })
 
     // signal the multipart that the underlying stream has drained and that