feat: avoid doing multiple RPC requests for files.add, fixes #522

Author: daviddias

src/files/add.js

@@ -2,7 +2,7 @@
 
 const isStream = require('is-stream')
 const promisify = require('promisify-es6')
-const DAGNodeStream = require('../utils/dagnode-stream')
+const converter = require('../utils/converter')
 
 module.exports = (send) => {
   return promisify((files, opts, callback) => {
@@ -23,8 +23,6 @@ module.exports = (send) => {
 
     const request = { path: 'add', files: files, qs: opts }
 
-    // Transform the response stream to DAGNode values
-    const transform = (res, callback) => DAGNodeStream.streamToValue(send, res, callback)
-    send.andTransform(request, transform, callback)
+    send.andTransform(request, converter, callback)
   })
 }

src/util/fs-add.js

@@ -2,7 +2,7 @@
 
 const isNode = require('detect-node')
 const promisify = require('promisify-es6')
-const DAGNodeStream = require('../utils/dagnode-stream')
+const converter = require('../utils/converter')
 const moduleConfig = require('../utils/module-config')
 
 module.exports = (arg) => {
@@ -31,14 +31,7 @@ module.exports = (arg) => {
       return callback(new Error('"path" must be a string'))
     }
 
-    const request = {
-      path: 'add',
-      qs: opts,
-      files: path
-    }
-
-    // Transform the response stream to DAGNode values
-    const transform = (res, callback) => DAGNodeStream.streamToValue(send, res, callback)
-    send.andTransform(request, transform, callback)
+    const request = { path: 'add', files: path, qs: opts }
+    send.andTransform(request, converter, callback)
   })
 }

src/util/url-add.js

@@ -4,7 +4,7 @@ const promisify = require('promisify-es6')
 const once = require('once')
 const parseUrl = require('url').parse
 const request = require('../utils/request')
-const DAGNodeStream = require('../utils/dagnode-stream')
+const converter = require('../utils/converter')
 const moduleConfig = require('../utils/module-config')
 
 module.exports = (arg) => {
@@ -25,12 +25,12 @@ module.exports = (arg) => {
       opts = {}
     }
 
+    callback = once(callback)
+
     if (!validUrl(url)) {
       return callback(new Error('"url" param must be an http(s) url'))
     }
 
-    callback = once(callback)
-
     requestWithRedirect(url, opts, send, callback)
   })
 }
@@ -52,14 +52,9 @@ const requestWithRedirect = (url, opts, send, callback) => {
       }
       requestWithRedirect(redirection, opts, send, callback)
     } else {
-      const params = {
-        path: 'add',
-        qs: opts,
-        files: res
-      }
-      // Transform the response stream to DAGNode values
-      const transform = (res, callback) => DAGNodeStream.streamToValue(send, res, callback)
-      send.andTransform(params, transform, callback)
+      const request = { path: 'add', files: res, qs: opts }
+
+      send.andTransform(request, converter, callback)
     }
   }).end()
 }

src/utils/get-dagnode.js renamed to src/utils/aaa

@@ -3,18 +3,18 @@
 const pump = require('pump')
 const TransformStream = require('readable-stream').Transform
 const streamToValue = require('./stream-to-value')
-const getDagNode = require('./get-dagnode')
 
 /*
   Transforms a stream of {Name, Hash} objects to include size
   of the DAG object.
 
-  Usage: inputStream.pipe(DAGNodeStream({ send: send }))
+  Usage: inputStream.pipe(new Converter())
 
   Input object format:
   {
     Name: '/path/to/file/foo.txt',
     Hash: 'Qma4hjFTnCasJ8PVp3mZbZK5g2vGDT4LByLJ7m8ciyRFZP'
+    Size: '20'
   }
 
   Output object format:
@@ -24,38 +24,34 @@ const getDagNode = require('./get-dagnode')
     size: 20
   }
 */
-class DAGNodeStream extends TransformStream {
+class ConverterStream extends TransformStream {
   constructor (options) {
     const opts = Object.assign(options || {}, { objectMode: true })
     super(opts)
-    this._send = opts.send
   }
 
-  static streamToValue (send, inputStream, callback) {
-    const outputStream = pump(inputStream, new DAGNodeStream({ send: send }), (err) => {
-      if (err) {
-        callback(err)
-      }
+  _transform (obj, enc, callback) {
+    this.push({
+      path: obj.Name,
+      hash: obj.Hash,
+      size: parseInt(obj.Size, 10)
     })
-    streamToValue(outputStream, callback)
+
+    callback(null)
   }
+}
 
-  _transform (obj, enc, callback) {
-    getDagNode(this._send, obj.Hash, (err, node) => {
+function converter (inputStream, callback) {
+  const outputStream = pump(
+    inputStream,
+    new ConverterStream(),
+    (err) => {
       if (err) {
-        return callback(err)
-      }
-
-      const result = {
-        path: obj.Name,
-        hash: obj.Hash,
-        size: node.size
+        callback(err)
       }
-
-      this.push(result)
-      callback(null)
     })
-  }
+
+  streamToValue(outputStream, callback)
 }
 
-module.exports = DAGNodeStream
+module.exports = converter

src/utils/dagnode-stream.js renamed to src/utils/converter.js

@@ -6,9 +6,9 @@ const concat = require('concat-stream')
 /*
   Concatenate a stream to a single value.
 */
-function streamToValue (res, callback) {
+function streamToValue (response, callback) {
   pump(
-    res,
+    response,
     concat((data) => callback(null, data)),
     (err) => {
       if (err) {