feat(breaking change): use stream on stats.bw (#686)

* feat(breaking change): use stream on stats.bw

* add some type checking

* readable stream and pull stream on bandwidth stats

* fix bw pull stream

* Bump interface-ipfs-core version

Author: hacdias

package.json

@@ -24,6 +24,7 @@
   },
   "dependencies": {
     "async": "^2.6.0",
+    "big.js": "^5.0.3",
     "bs58": "^4.0.1",
     "cids": "~0.5.2",
     "concat-stream": "^1.6.0",
@@ -68,8 +69,8 @@
     "eslint-plugin-react": "^7.5.1",
     "go-ipfs-dep": "^0.4.13",
     "gulp": "^3.9.1",
-    "interface-ipfs-core": "~0.43.0",
     "hapi": "^16.6.2",
+    "interface-ipfs-core": "~0.47.0",
     "ipfsd-ctl": "~0.27.0",
     "pre-commit": "^1.2.2",
     "socket.io": "^2.0.4",

src/bitswap/stat.js

@@ -1,18 +1,19 @@
 'use strict'
 
 const promisify = require('promisify-es6')
+const Big = require('big.js')
 
 const transform = function (res, callback) {
   callback(null, {
     provideBufLen: res.ProvideBufLen,
-    wantlist: res.Wantlist,
-    peers: res.Peers,
-    blocksReceived: res.BlocksReceived,
-    dataReceived: res.DataReceived,
-    blocksSent: res.BlocksSent,
-    dataSent: res.DataSent,
-    dupBlksReceived: res.DupBlksReceived,
-    dupDataReceived: res.DupDataReceived
+    wantlist: res.Wantlist || [],
+    peers: res.Peers || [],
+    blocksReceived: new Big(res.BlocksReceived),
+    dataReceived: new Big(res.DataReceived),
+    blocksSent: new Big(res.BlocksSent),
+    dataSent: new Big(res.DataSent),
+    dupBlksReceived: new Big(res.DupBlksReceived),
+    dupDataReceived: new Big(res.DupDataReceived)
   })
 }
 

src/repo/stat.js

@@ -1,14 +1,15 @@
 'use strict'
 
 const promisify = require('promisify-es6')
+const Big = require('big.js')
 
 const transform = function (res, callback) {
   callback(null, {
-    numObjects: res.NumObjects,
-    repoSize: res.RepoSize,
+    numObjects: new Big(res.NumObjects),
+    repoSize: new Big(res.RepoSize),
     repoPath: res.RepoPath,
     version: res.Version,
-    storageMax: res.StorageMax
+    storageMax: new Big(res.StorageMax)
   })
 }
 

src/stats/bitswap.js

@@ -1,18 +1,19 @@
 'use strict'
 
 const promisify = require('promisify-es6')
+const Big = require('big.js')
 
 const transform = function (res, callback) {
   callback(null, {
     provideBufLen: res.ProvideBufLen,
-    wantlist: res.Wantlist,
-    peers: res.Peers,
-    blocksReceived: res.BlocksReceived,
-    dataReceived: res.DataReceived,
-    blocksSent: res.BlocksSent,
-    dataSent: res.DataSent,
-    dupBlksReceived: res.DupBlksReceived,
-    dupDataReceived: res.DupDataReceived
+    wantlist: res.Wantlist || [],
+    peers: res.Peers || [],
+    blocksReceived: new Big(res.BlocksReceived),
+    dataReceived: new Big(res.DataReceived),
+    blocksSent: new Big(res.BlocksSent),
+    dataSent: new Big(res.DataSent),
+    dupBlksReceived: new Big(res.DupBlksReceived),
+    dupDataReceived: new Big(res.DupDataReceived)
   })
 }
 

src/stats/bw-pull-stream.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const toPull = require('stream-to-pull-stream')
+const pull = require('pull-stream')
+const transformChunk = require('./bw-util')
+const deferred = require('pull-defer')
+
+module.exports = (send) => {
+  return (hash, opts) => {
+    opts = opts || {}
+
+    const p = deferred.source()
+
+    send({
+      path: 'stats/bw',
+      qs: opts
+    }, (err, stream) => {
+      if (err) {
+        return p.end(err)
+      }
+
+      p.resolve(pull(
+        toPull.source(stream),
+        pull.map(transformChunk)
+      ))
+    })
+
+    return p
+  }
+}

src/stats/bw-readable-stream.js

@@ -0,0 +1,31 @@
+'use strict'
+
+const Stream = require('readable-stream')
+const pump = require('pump')
+const transformChunk = require('./bw-util')
+
+module.exports = (send) => {
+  return (hash, opts) => {
+    opts = opts || {}
+
+    const pt = new Stream.Transform({
+      objectMode: true,
+      transform (chunk, encoding, cb) {
+        cb(null, transformChunk(chunk))
+      }
+    })
+
+    send({
+      path: 'stats/bw',
+      qs: opts
+    }, (err, stream) => {
+      if (err) {
+        return pt.destroy(err)
+      }
+
+      pump(stream, pt)
+    })
+
+    return pt
+  }
+}

src/stats/bw-util.js

@@ -0,0 +1,12 @@
+'use strict'
+
+const Big = require('big.js')
+
+module.exports = (chunk) => {
+  return {
+    totalIn: new Big(chunk.TotalIn),
+    totalOut: new Big(chunk.TotalOut),
+    rateIn: new Big(chunk.RateIn),
+    rateOut: new Big(chunk.RateOut)
+  }
+}

src/stats/bw.js

@@ -2,19 +2,15 @@
 
 const promisify = require('promisify-es6')
 const streamToValue = require('../utils/stream-to-value')
+const transformChunk = require('./bw-util')
 
-const transform = function (res, callback) {
-  streamToValue(res, (err, data) => {
+const transform = (res, callback) => {
+  return streamToValue(res, (err, data) => {
     if (err) {
       return callback(err)
     }
 
-    callback(null, {
-      totalIn: data[0].TotalIn,
-      totalOut: data[0].TotalOut,
-      rateIn: data[0].RateIn,
-      rateOut: data[0].RateOut
-    })
+    callback(null, transformChunk(data[0]))
   })
 }
 

src/stats/index.js

@@ -8,6 +8,8 @@ module.exports = (arg) => {
   return {
     bitswap: require('./bitswap')(send),
     bw: require('./bw')(send),
+    bwReadableStream: require('./bw-readable-stream')(send),
+    bwPullStream: require('./bw-pull-stream')(send),
     repo: require('./repo')(send)
   }
 }

src/stats/repo.js

@@ -1,14 +1,15 @@
 'use strict'
 
 const promisify = require('promisify-es6')
+const Big = require('big.js')
 
 const transform = function (res, callback) {
   callback(null, {
-    numObjects: res.NumObjects,
-    repoSize: res.RepoSize,
+    numObjects: new Big(res.NumObjects),
+    repoSize: new Big(res.RepoSize),
     repoPath: res.RepoPath,
     version: res.Version,
-    storageMax: res.StorageMax
+    storageMax: new Big(res.StorageMax)
   })
 }