refactor: update deps and remove highland

now using pull-streams instead

Fixes #24

Author: dignifiedquire

package.json

@@ -34,39 +34,40 @@
   "homepage": "https://github.com/ipfs/js-ipfs-bitswap#readme",
   "devDependencies": {
     "abstract-blob-store": "^3.2.0",
-    "aegir": "^3.2.0",
+    "aegir": "^6.0.1",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
     "fs-blob-store": "^5.2.1",
     "idb-plus-blob-store": "^1.1.2",
     "ipfs-repo": "^0.8.0",
-    "libp2p-ipfs": "^0.12.0",
-    "lodash": "^4.13.1",
+    "libp2p-ipfs": "^0.12.1",
+    "lodash": "^4.15.0",
     "multiaddr": "^2.0.3",
     "ncp": "^2.0.0",
     "peer-book": "^0.3.0",
     "peer-id": "^0.7.0",
     "peer-info": "^0.7.1",
-    "rimraf": "^2.5.2",
+    "rimraf": "^2.5.4",
     "safe-buffer": "^5.0.1"
   },
   "dependencies": {
-    "async": "^2.0.0-rc.5",
+    "async": "^2.0.1",
     "bl": "^1.1.2",
     "debug": "^2.2.0",
     "heap": "^0.2.6",
-    "highland": "^3.0.0-beta.1",
     "ipfs-block": "^0.3.0",
     "length-prefixed-stream": "^1.5.0",
-    "lodash.isequalwith": "^4.2.0",
+    "lodash.isequalwith": "^4.4.0",
     "lodash.isundefined": "^3.0.1",
     "multihashes": "^0.2.2",
-    "protocol-buffers": "^3.1.6"
+    "protocol-buffers": "^3.1.6",
+    "pull-generate": "^2.2.0",
+    "pull-stream": "^3.4.3"
   },
   "contributors": [
     "David Dias <[email protected]>",
     "Richard Littauer <[email protected]>",
     "Stephen Whitmore <[email protected]>",
     "dignifiedquire <[email protected]>"
   ]
-}
+}

src/decision/engine.js

@@ -1,9 +1,10 @@
 'use strict'
 
 const debug = require('debug')
-const _ = require('highland')
 const async = require('async')
 const mh = require('multihashes')
+const pull = require('pull-stream')
+const generate = require('pull-generate')
 
 const log = debug('bitswap:engine')
 log.error = debug('bitswap:engine:error')
@@ -45,34 +46,38 @@ module.exports = class Engine {
   _outbox () {
     if (!this._running) return
 
-    const doIt = (cb) => {
-      _((push, next) => {
-        if (!this._running) return push(null, _.nil)
+    const doIt = (cb) => pull(
+      generate(null, (state, cb) => {
+        if (!this._running) {
+          return cb(true)
+        }
+
         const nextTask = this.peerRequestQueue.pop()
 
-        if (!nextTask) return push(null, _.nil)
+        if (!nextTask) {
+          return cb(true)
+        }
 
         this.datastore.get(nextTask.entry.key, (err, block) => {
           if (err || !block) {
             nextTask.done()
-          } else {
-            push(null, {
-              peer: nextTask.target,
-              block: block,
-              sent: () => {
-                nextTask.done()
-              }
-            })
+            return cb()
           }
 
-          next()
+          cb(null, {
+            peer: nextTask.target,
+            block: block,
+            sent: () => {
+              nextTask.done()
+            }
+          })
         })
-      })
-        .flatMap((envelope) => {
-          return _.wrapCallback(this._sendBlock.bind(this))(envelope)
-        })
-        .done(cb)
-    }
+      }),
+      pull.asyncMap((envelope, cb) => {
+        this._sendBlock(envelope, cb)
+      }),
+      pull.onEnd(cb)
+    )
 
     if (!this._timer) {
       this._timer = setTimeout(() => {

src/wantmanager/index.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const debug = require('debug')
-const _ = require('highland')
+const pull = require('pull-stream')
 
 const Message = require('../message')
 const Wantlist = require('../wantlist')
@@ -25,12 +25,13 @@ module.exports = class Wantmanager {
 
   _addEntries (keys, cancel, force) {
     let i = -1
-    _(keys)
-      .map((key) => {
+    pull(
+      pull.values(keys),
+      pull.map((key) => {
         i++
         return new Message.Entry(key, cs.kMaxPriority - i, cancel)
-      })
-      .tap((e) => {
+      }),
+      pull.through((e) => {
         // add changes to our wantlist
         if (e.cancel) {
           if (force) {
@@ -41,13 +42,15 @@ module.exports = class Wantmanager {
         } else {
           this.wl.add(e.key, e.priority)
         }
-      })
-      .toArray((entries) => {
+      }),
+      pull.collect((err, entries) => {
+        if (err) throw err
         // broadcast changes
         for (let p of this.peers.values()) {
           p.addEntries(entries, false)
         }
       })
+    )
   }
 
   _startPeerHandler (peerId) {

test/decision/engine-test.js

@@ -161,12 +161,16 @@ module.exports = (repo) => {
               const cancels = testcase[1]
               const keeps = _.difference(set, cancels)
 
-              const network = mockNetwork(keeps.length, (res) => {
-                const msgs = _.flatten(res.messages.map(
-                  (m) => Array.from(m[1].blocks.values())
-                    .map((b) => b.data.toString())
-                ))
+              const messageToString = (m) => {
+                return Array.from(m[1].blocks.values())
+                  .map((b) => b.data.toString())
+              }
+              const stringifyMessages = (messages) => {
+                return _.flatten(messages.map(messageToString))
+              }
 
+              const network = mockNetwork(keeps.length, (res) => {
+                const msgs = stringifyMessages(res.messages)
                 expect(msgs).to.be.eql(keeps)
                 innerCb()
               })

test/index-test.js

@@ -1,4 +1,5 @@
 /* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 8]*/
 'use strict'
 
 const async = require('async')
@@ -61,18 +62,15 @@ module.exports = (repo) => {
           expect(bs.blocksRecvd).to.be.eql(2)
           expect(bs.dupBlocksRecvd).to.be.eql(0)
 
-          async.parallel([
-            (cb) => store.get(b1.key, (err, res) => {
-              if (err) cb(err)
-              expect(res).to.be.eql(b1)
-              cb()
-            }),
-            (cb) => store.get(b1.key, (err, res) => {
-              if (err) return cb(err)
-              expect(res).to.be.eql(b1)
-              cb()
-            })
-          ], done)
+          async.map([b1, b1],
+            (val, cb) => store.get(val.key, cb),
+            (err, res) => {
+              if (err) return done(err)
+
+              expect(res).to.be.eql([b1, b1])
+              done()
+            }
+          )
         })
       })
 

test/network/gen-bitswap-network.node.js

@@ -1,4 +1,5 @@
 /* eslint-env mocha */
+/* eslint max-nested-callbacks: ["error", 8]*/
 'use strict'
 
 const expect = require('chai').expect