Bitswap Quest

.travis.yml

@@ -14,12 +14,21 @@ script:
   - npm test
   - npm run coverage
 
-addons:
-  firefox: 'latest'
 
 before_script:
   - export DISPLAY=:99.0
   - sh -e /etc/init.d/xvfb start
 
 after_success:
   - npm run coverage-publish
+
+env:
+  - CXX=g++-4.8
+
+addons:
+  firefox: 'latest'
+  apt:
+    sources:
+      - ubuntu-toolchain-r-test
+    packages:
+      - g++-4.8

gulpfile.js

@@ -0,0 +1,41 @@
+'use strict'
+
+const gulp = require('gulp')
+const parallel = require('run-parallel')
+const series = require('run-series')
+const createTempNode = require('./test/utils/temp-node')
+const API = require('./src/http-api')
+
+let nodes = []
+
+function startNode (num, done) {
+  createTempNode(num, (err, node) => {
+    if (err) throw err
+
+    const api = new API(node.repo.path())
+    nodes.push(api)
+    api.start(done)
+  })
+}
+
+gulp.task('libnode:start', (done) => {
+  nodes = []
+  parallel([
+    (cb) => startNode(7, cb),
+    (cb) => startNode(8, cb),
+    (cb) => startNode(9, cb)
+  ], done)
+})
+
+gulp.task('libnode:stop', (done) => {
+  series(nodes.map((node) => (cb) => {
+    setTimeout(() => node.stop(cb), 500)
+  }), done)
+})
+
+gulp.task('test:browser:before', ['libnode:start'])
+gulp.task('test:node:before', ['libnode:start'])
+gulp.task('test:browser:after', ['libnode:stop'])
+gulp.task('test:node:after', ['libnode:stop'])
+
+require('aegir/gulp')(gulp)

package.json

@@ -9,14 +9,14 @@
   "jsnext:main": "src/core/index.js",
   "scripts": {
     "lint": "aegir-lint",
-    "coverage": "aegir-coverage",
-    "test": "aegir-test",
-    "test:node": "aegir-test node",
-    "test:browser": "aegir-test browser",
-    "build": "aegir-build",
-    "release": "aegir-release",
-    "release:minor": "aegir-release --type minor",
-    "release:major": "aegir-release --type major",
+    "coverage": "gulp coverage",
+    "test": "gulp test",
+    "test:node": "gulp test:node",
+    "test:browser": "gulp test:browser",
+    "build": "gulp build",
+    "release": "gulp release",
+    "release:minor": "gulp release --type minor",
+    "release:major": "gulp release --type major",
     "coverage-publish": "aegir-coverage publish"
   },
   "pre-commit": [
@@ -38,12 +38,13 @@
   "homepage": "https://github.com/ipfs/js-ipfs#readme",
   "devDependencies": {
     "aegir": "^3.0.1",
-    "async": "^2.0.0-rc.4",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
     "expose-loader": "^0.7.1",
     "form-data": "^1.0.0-rc3",
+    "gulp": "^3.9.1",
     "idb-plus-blob-store": "^1.1.2",
+    "libp2p-ipfs-browser": "^0.2.0",
     "lodash": "^4.11.2",
     "mocha": "^2.4.5",
     "ncp": "^2.0.0",
@@ -63,6 +64,7 @@
     "glob": "^7.0.3",
     "hapi": "^13.3.0",
     "ipfs-api": "^3.0.2",
+    "ipfs-bitswap": "^0.2.0",
     "ipfs-block": "^0.3.0",
     "ipfs-block-service": "^0.4.0",
     "ipfs-merkle-dag": "^0.5.0",
@@ -71,29 +73,32 @@
     "ipfs-unixfs-engine": "^0.6.1",
     "joi": "^8.0.5",
     "libp2p-ipfs": "^0.3.3",
-    "libp2p-swarm": "^0.12.5",
+    "libp2p-swarm": "^0.12.11",
     "lodash.get": "^4.2.1",
     "lodash.set": "^4.1.0",
     "multiaddr": "^1.4.1",
     "path-exists": "^3.0.0",
-    "peer-book": "0.1.0",
+    "peer-book": "^0.1.1",
     "peer-id": "^0.6.6",
     "peer-info": "^0.6.2",
     "readable-stream": "1.1.13",
     "ronin": "^0.3.11",
+    "run-parallel": "^1.1.6",
+    "run-parallel-limit": "^1.0.3",
+    "run-series": "^1.1.4",
+    "run-waterfall": "^1.1.3",
     "temp": "^0.8.3"
   },
   "aegir": {
     "webpack": {
       "resolve": {
         "alias": {
-          "node-forge": "../../../node_modules/peer-id/vendor/forge.bundle.js"
+          "node-forge": "../../../node_modules/peer-id/vendor/forge.bundle.js",
+          "libp2p-ipfs": "libp2p-ipfs-browser"
         }
       },
       "externals": {
         "fs": "{}",
-        "ipfs-data-importing": "{ import: {} }",
-        "libp2p-ipfs": "{}",
         "mkdirp": "{}"
       }
     }

src/cli/commands/bitswap/stat.js

@@ -0,0 +1,38 @@
+'use strict'
+
+const Command = require('ronin').Command
+const utils = require('../../utils')
+
+module.exports = Command.extend({
+  desc: 'Show some diagnostic information on the bitswap agent.',
+
+  options: {
+  },
+
+  run: () => {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.bitswap.stat((err, stats) => {
+        if (err) {
+          throw err
+        }
+
+        stats.Wantlist = stats.Wantlist || []
+        stats.Peers = stats.Peers || []
+
+        console.log(`
+bitswap status
+  blocks received: ${stats.BlocksReceived}
+  dup blocks received: ${stats.DupBlksReceived}
+  dup data received: ${stats.DupDataReceived}B
+  wantlist [${stats.Wantlist.length} keys]
+    ${stats.Wantlist.join('\n    ')}
+  partners [${stats.Peers.length}]
+    ${stats.Peers.join('\n    ')}`)
+      })
+    })
+  }
+})

src/cli/commands/bitswap/unwant.js

@@ -0,0 +1,24 @@
+'use strict'
+
+const Command = require('ronin').Command
+const utils = require('../../utils')
+
+module.exports = Command.extend({
+  desc: 'Remove a given block from your wantlist.',
+
+  options: {
+    key: {
+      required: true
+    }
+  },
+
+  run: (key) => {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      // TODO: implement
+    })
+  }
+})

src/cli/commands/bitswap/wantlist.js

@@ -0,0 +1,27 @@
+'use strict'
+
+const Command = require('ronin').Command
+const utils = require('../../utils')
+
+module.exports = Command.extend({
+  desc: 'Print out all blocks currently on the bitswap wantlist for the local peer.',
+
+  options: {
+  },
+
+  run: () => {
+    utils.getIPFS((err, ipfs) => {
+      if (err) {
+        throw err
+      }
+
+      ipfs.bitswap.wantlist((err, res) => {
+        if (err) {
+          throw err
+        }
+
+        res.Keys.forEach((k) => console.log(k))
+      })
+    })
+  }
+})

src/cli/commands/config/edit.js

@@ -4,7 +4,7 @@ const Command = require('ronin').Command
 const spawn = require('child_process').spawn
 const fs = require('fs')
 const temp = require('temp')
-const async = require('async')
+const waterfall = require('run-waterfall')
 const debug = require('debug')
 const log = debug('cli:config')
 log.error = debug('cli:config:error')
@@ -99,7 +99,7 @@ module.exports = Command.extend({
         })
       }
 
-      async.waterfall([
+      waterfall([
         getConfig,
         saveTempConfig,
         openEditor,

src/cli/commands/files/add.js

@@ -7,7 +7,7 @@ const log = debug('cli:version')
 log.error = debug('cli:version:error')
 const bs58 = require('bs58')
 const fs = require('fs')
-const async = require('async')
+const parallelLimit = require('run-parallel-limit')
 const path = require('path')
 const glob = require('glob')
 
@@ -69,15 +69,15 @@ module.exports = Command.extend({
         })
         if (res.length !== 0) {
           const index = inPath.lastIndexOf('/')
-          async.eachLimit(res, 10, (element, callback) => {
+          parallelLimit(res.map((element) => (callback) => {
             if (!fs.statSync(element).isDirectory()) {
               i.write({
                 path: element.substring(index + 1, element.length),
                 stream: fs.createReadStream(element)
               })
             }
             callback()
-          }, (err) => {
+          }), 10, (err) => {
             if (err) {
               throw err
             }

src/core/index.js

@@ -7,6 +7,9 @@ const PeerBook = require('peer-book')
 
 const defaultRepo = require('./default-repo')
 
+const goOnline = require('./ipfs/go-online')
+const goOffline = require('./ipfs/go-offline')
+const isOnline = require('./ipfs/is-online')
 const load = require('./ipfs/load')
 const version = require('./ipfs/version')
 const id = require('./ipfs/id')
@@ -18,6 +21,7 @@ const block = require('./ipfs/block')
 const object = require('./ipfs/object')
 const libp2p = require('./ipfs/libp2p')
 const files = require('./ipfs/files')
+const bitswap = require('./ipfs/bitswap')
 
 exports = module.exports = IPFS
 
@@ -32,12 +36,16 @@ function IPFS (repoInstance) {
   }
 
   this._repo = repoInstance
-  this._blockS = new BlockService(this._repo)
-  this._dagS = new DAGService(this._blockS)
   this._peerInfoBook = new PeerBook()
   this._peerInfo = null
   this._libp2pNode = null
+  this._bitswap = null
+  this._blockS = new BlockService(this._repo)
+  this._dagS = new DAGService(this._blockS)
 
+  this.goOnline = goOnline(this)
+  this.goOffline = goOffline(this)
+  this.isOnline = isOnline(this)
   this.load = load(this)
   this.version = version(this)
   this.id = id(this)
@@ -49,4 +57,5 @@ function IPFS (repoInstance) {
   this.object = object(this)
   this.libp2p = libp2p(this)
   this.files = files(this)
+  this.bitswap = bitswap(this)
 }

src/core/ipfs/bitswap.js

@@ -0,0 +1,42 @@
+'use strict'
+
+const bs58 = require('bs58')
+
+const OFFLINE_ERROR = require('../utils').OFFLINE_ERROR
+
+function formatWantlist (list) {
+  return Array.from(list).map((el) => {
+    return bs58.encode(new Buffer(el[0], 'hex'))
+  })
+}
+
+module.exports = function bitswap (self) {
+  return {
+    wantlist: () => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      const list = self._bitswap.getWantlist()
+      return formatWantlist(list)
+    },
+    stat: () => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      const stats = self._bitswap.stat()
+      stats.wantlist = formatWantlist(stats.wantlist)
+      stats.peers = stats.peers.map((id) => id.toB58String())
+
+      return stats
+    },
+    unwant: (key) => {
+      if (!self.isOnline()) {
+        throw OFFLINE_ERROR
+      }
+
+      // TODO: implement when https://github.com/ipfs/js-ipfs-bitswap/pull/10 is merged
+    }
+  }
+}

src/core/ipfs/go-offline.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const debug = require('debug')
+const log = debug('core:offline')
+
+module.exports = function goOffline (self) {
+  return (cb) => {
+    self._blockS.goOffline()
+    self._bitswap.stop()
+    self.libp2p.stop((err) => {
+      if (err) {
+        log('Error trying to go offline', err)
+      }
+      cb()
+    })
+  }
+}