Use stream-to-value in Block command instead of bl

Author: haadcode

package.json

@@ -23,7 +23,6 @@
   },
   "dependencies": {
     "async": "^2.1.4",
-    "bl": "^1.1.2",
     "bs58": "^3.1.0",
     "concat-stream": "^1.5.2",
     "detect-node": "^2.0.3",
@@ -43,6 +42,7 @@
     "peer-id": "^0.8.1",
     "peer-info": "^0.8.1",
     "promisify-es6": "^1.0.2",
+    "pump": "^1.0.1",
     "qs": "^6.3.0",
     "readable-stream": "1.1.14",
     "stream-http": "^2.5.0",
@@ -117,4 +117,4 @@
     "url": "https://github.com/ipfs/js-ipfs-api/issues"
   },
   "homepage": "https://github.com/ipfs/js-ipfs-api"
-}
+}

src/api/block.js

@@ -1,10 +1,10 @@
 'use strict'
 
 const promisify = require('promisify-es6')
-const bl = require('bl')
 const Block = require('ipfs-block')
 const multihash = require('multihashes')
 const CID = require('cids')
+const streamToValue = require('../stream-to-value')
 
 module.exports = (send) => {
   return {
@@ -21,25 +21,27 @@ module.exports = (send) => {
         opts = {}
       }
 
-      return send({
-        path: 'block/get',
-        args: args,
-        qs: opts
-      }, (err, res) => {
-        if (err) {
-          return callback(err)
-        }
+      // Transform the response from Buffer or a Stream to a Block
+      const transform = (res, callback) => {
         if (Buffer.isBuffer(res)) {
           callback(null, new Block(res))
         } else {
-          res.pipe(bl((err, data) => {
+          streamToValue(res, (err, data) => {
             if (err) {
               return callback(err)
             }
             callback(null, new Block(data))
-          }))
+          })
         }
-      })
+      }
+
+      const request = {
+        path: 'block/get',
+        args: args,
+        qs: opts
+      }
+
+      send.andTransform(request, transform, callback)
     }),
     stat: promisify((args, opts, callback) => {
       // TODO this needs to be adjusted with the new go-ipfs http-api
@@ -51,19 +53,22 @@ module.exports = (send) => {
         callback = opts
         opts = {}
       }
-      return send({
+
+      const request = {
         path: 'block/stat',
         args: args,
         qs: opts
-      }, (err, stats) => {
-        if (err) {
-          return callback(err)
-        }
+      }
+
+      // Transform the response from { Key, Size } objects to { key, size } objects
+      const transform = (stats, callback) => {
         callback(null, {
           key: stats.Key,
           size: stats.Size
         })
-      })
+      }
+
+      send.andTransform(request, transform, callback)
     }),
     put: promisify((block, cid, callback) => {
       // TODO this needs to be adjusted with the new go-ipfs http-api
@@ -81,15 +86,15 @@ module.exports = (send) => {
         block = block.data
       }
 
-      return send({
+      const request = {
         path: 'block/put',
         files: block
-      }, (err, blockInfo) => {
-        if (err) {
-          return callback(err)
-        }
-        callback(null, new Block(block))
-      })
+      }
+
+      // Transform the response to a Block
+      const transform = (blockInfo, callback) => callback(null, new Block(block))
+
+      send.andTransform(request, transform, callback)
     })
   }
 }

src/api/log.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const pump = require('pump')
 const ndjson = require('ndjson')
 const promisify = require('promisify-es6')
 
@@ -12,7 +13,8 @@ module.exports = (send) => {
         if (err) {
           return callback(err)
         }
-        callback(null, response.pipe(ndjson.parse()))
+        const outputStream = pump(response, ndjson.parse())
+        callback(null, outputStream)
       })
     })
   }

src/api/object.js

@@ -5,7 +5,7 @@ const DAGNode = dagPB.DAGNode
 const DAGLink = dagPB.DAGLink
 const promisify = require('promisify-es6')
 const bs58 = require('bs58')
-const bl = require('bl')
+const streamToValue = require('../stream-to-value')
 const cleanMultihash = require('../clean-multihash')
 const LRU = require('lru-cache')
 const lruOptions = {
@@ -188,7 +188,7 @@ module.exports = (send) => {
         }
 
         if (typeof result.pipe === 'function') {
-          result.pipe(bl(callback))
+          streamToValue(result, callback)
         } else {
           callback(null, result)
         }

src/dagnode-stream.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const pump = require('pump')
 const TransformStream = require('readable-stream').Transform
 const streamToValue = require('./stream-to-value')
 const getDagNode = require('./get-dagnode')
@@ -31,7 +32,9 @@ class DAGNodeStream extends TransformStream {
   }
 
   static streamToValue (send, inputStream, callback) {
-    const outputStream = inputStream.pipe(new DAGNodeStream({ send: send }))
+    const outputStream = pump(inputStream, new DAGNodeStream({ send: send }), (err) => {
+      if (err) callback(err)
+    })
     streamToValue(outputStream, callback)
   }
 

src/request-api.js

@@ -3,6 +3,7 @@
 const Qs = require('qs')
 const isNode = require('detect-node')
 const ndjson = require('ndjson')
+const pump = require('pump')
 const once = require('once')
 const getFilesStream = require('./get-files-stream')
 const streamToValue = require('./stream-to-value')
@@ -43,7 +44,8 @@ function onRes (buffer, cb) {
 
     // Return a stream of JSON objects
     if (chunkedObjects && isJson) {
-      return cb(null, res.pipe(ndjson.parse()))
+      const outputStream = pump(res, ndjson.parse())
+      return cb(null, outputStream)
     }
 
     // Return a JSON object

src/stream-to-value.js

@@ -1,12 +1,16 @@
 'use strict'
 
+const pump = require('pump')
 const concat = require('concat-stream')
 
 /*
   Concatenate a stream to a single value.
 */
 function streamToValue (res, callback) {
-  res.pipe(concat((data) => callback(null, data)))
+  const done = (data) => callback(null, data)
+  pump(res, concat(done), (err) => {
+    if (err) callback(err)
+  })
 }
 
 module.exports = streamToValue

src/tar-stream-to-objects.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const pump = require('pump')
 const tar = require('tar-stream')
 const ReadableStream = require('readable-stream').Readable
 
@@ -9,7 +10,7 @@ class ObjectsStreams extends ReadableStream {
     super(opts)
   }
 
-  _read ()  {}
+  _read () {}
 }
 
 /*
@@ -20,9 +21,9 @@ class ObjectsStreams extends ReadableStream {
 */
 const TarStreamToObjects = (inputStream, callback) => {
   let outputStream = new ObjectsStreams()
+  let extractStream = tar.extract()
 
-  inputStream
-    .pipe(tar.extract())
+  extractStream
     .on('entry', (header, stream, next) => {
       stream.on('end', next)
 
@@ -40,6 +41,7 @@ const TarStreamToObjects = (inputStream, callback) => {
     })
     .on('finish', () => outputStream.push(null))
 
+  pump(inputStream, extractStream)
   callback(null, outputStream)
 }