wip: message and network support (need to commit before the plane)

package.json

@@ -35,14 +35,14 @@
   },
   "homepage": "https://github.com/ipfs/js-ipfs-bitswap#readme",
   "devDependencies": {
-    "aegir": "9.2.1",
+    "aegir": "9.3.0",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
     "fs-pull-blob-store": "^0.4.1",
     "idb-pull-blob-store": "^0.5.1",
     "interface-pull-blob-store": "^0.6.0",
-    "ipfs-repo": "^0.11.1",
-    "libp2p-ipfs-nodejs": "^0.16.4",
+    "ipfs-repo": "^0.11.2",
+    "libp2p-ipfs-nodejs": "^0.17.0",
     "lodash": "^4.17.2",
     "multiaddr": "^2.1.1",
     "ncp": "^2.0.0",
@@ -54,19 +54,20 @@
   },
   "dependencies": {
     "async": "^2.1.4",
-    "cids": "^0.3.4",
-    "debug": "^2.3.3",
+    "cids": "^0.3.5",
+    "debug": "^2.4.4",
     "heap": "^0.2.6",
-    "ipfs-block": "^0.5.1",
+    "ipfs-block": "^0.5.3",
     "lodash.debounce": "^4.0.8",
     "lodash.isequalwith": "^4.4.0",
-    "multihashes": "^0.3.0",
+    "multihashes": "^0.3.1",
     "protocol-buffers": "^3.2.1",
     "pull-defer": "^0.2.2",
     "pull-length-prefixed": "^1.2.0",
     "pull-paramap": "^1.2.1",
     "pull-pushable": "^2.0.1",
-    "pull-stream": "^3.5.0"
+    "pull-stream": "^3.5.0",
+    "varint-decoder": "^0.1.0"
   },
   "contributors": [
     "David Dias <daviddias.p@gmail.com>",

src/types/message/index.js

@@ -6,6 +6,7 @@ const isEqualWith = require('lodash.isequalwith')
 const assert = require('assert')
 const map = require('async/map')
 const CID = require('cids')
+const vd = require('varint-decoder')
 
 const pbm = protobuf(require('./message.proto'))
 const Entry = require('./entry')
@@ -81,6 +82,32 @@ class BitswapMessage {
    */
   serializeToBitswap110 () {
     // TODO
+    const msg = {
+      wantlist: {
+        entries: Array.from(this.wantlist.values()).map((entry) => {
+          return {
+            block: entry.cid.buffer, // cid
+            priority: Number(entry.priority),
+            cancel: Boolean(entry.cancel)
+          }
+        })
+      },
+      payload: []
+    }
+
+    if (this.full) {
+      msg.wantlist.full = true
+    }
+
+    this.blocks.forEach((cidStr, block) => {
+      const cid = new CID(cidStr)
+      msg.payload.push({
+        prefix: cid.prefix,
+        data: block.data
+      })
+    })
+
+    return pbm.Message.encode(msg)
   }
 
   equals (other) {
@@ -108,32 +135,72 @@ class BitswapMessage {
 }
 
 BitswapMessage.deserialize = (raw, callback) => {
-  const decoded = pbm.Message.decode(raw)
-  const msg = new BitswapMessage(decoded.wantlist.full)
+  let decoded
+  try {
+    decoded = pbm.Message.decode(raw)
+  } catch (err) {
+    return setImmediate(() => callback(err))
+  }
+
+  const isFull = (decoded.wantlist && decoded.wantlist.full) || false
+  const msg = new BitswapMessage(isFull)
 
-  decoded.wantlist.entries.forEach((entry) => {
-    // note: entry.block is the CID here
-    const cid = new CID(entry.block)
-    msg.addEntry(cid, entry.priority, entry.cancel)
-  })
+  if (decoded.wantlist) {
+    decoded.wantlist.entries.forEach((entry) => {
+      // note: entry.block is the CID here
+      const cid = new CID(entry.block)
+      msg.addEntry(cid, entry.priority, entry.cancel)
+    })
+  }
 
+  // Bitswap 1.0.0
   // decoded.blocks are just the byte arrays
-  map(decoded.blocks, (b, cb) => {
-    const block = new Block(b)
-    block.key((err, key) => {
+  if (decoded.blocks.length > 0) {
+    map(decoded.blocks, (b, cb) => {
+      const block = new Block(b)
+      block.key((err, key) => {
+        if (err) {
+          return cb(err)
+        }
+        const cid = new CID(key)
+        msg.addBlock(cid, block)
+        cb()
+      })
+    }, (err) => {
       if (err) {
-        return cb(err)
+        return callback(err)
       }
-      const cid = new CID(key)
-      msg.addBlock(cid, block)
-      cb()
+      callback(null, msg)
     })
-  }, (err) => {
-    if (err) {
-      return callback(err)
-    }
-    callback(null, msg)
-  })
+    return
+  }
+
+  // Bitswap 1.1.0
+  if (decoded.payload.length > 0) {
+    map(decoded.payload, (p, cb) => {
+      const values = vd(p.prefix)
+      const cidVersion = values[0]
+      const multicodec = values[1]
+      const hashAlg = values[2]
+      // const hashLen = values[3] // We haven't need to use this so far
+      const block = new Block(p.data)
+      block.key(hashAlg, (err, multihash) => {
+        if (err) {
+          return cb(err)
+        }
+        const cid = new CID(cidVersion, multicodec, multihash)
+        msg.addBlock(cid, block)
+        cb()
+      })
+    }, (err) => {
+      if (err) {
+        return callback(err)
+      }
+      callback(null, msg)
+    })
+    return
+  }
+  callback(null, msg)
 }
 
 BitswapMessage.Entry = Entry

src/types/message/message.proto.js

@@ -6,9 +6,7 @@ module.exports = `
   message Message {
     message Wantlist {
       message Entry {
-        // changed from string to bytes,
-        // because it makes a difference
-        // in JavaScript
+        // changed from string to bytes, it makes a difference in JavaScript
         optional bytes block = 1;      // the block cid (cidV0 in bitswap 1.0.0, cidV1 in bitswap 1.1.0)
         optional int32 priority = 2;    // the priority (normalized). default to 1
         optional bool cancel = 3;       // whether this revokes an entry

test/components/network/network.node.js

@@ -141,7 +141,7 @@ describe('network', () => {
     })
   })
 
-  it('._receiveMessage success', (done) => {
+  it('._receiveMessage success from Bitswap 1.0.0', (done) => {
     const msg = new Message(true)
     const b1 = blocks[0]
     const b2 = blocks[1]
@@ -184,6 +184,10 @@ describe('network', () => {
     })
   })
 
+  it.skip('._receiveMessage success from Bitswap 1.1.0', (done) => {})
+  it.skip('._sendMessage on Bitswap 1.0.0', (done) => {})
+  it.skip('._sendMessage on Bitswap 1.1.0', (done) => {})
+
   it('.sendMessage', (done) => {
     const msg = new Message(true)
     const b1 = blocks[0]

test/types/message.spec.js

@@ -8,6 +8,11 @@ const map = require('async/map')
 const pbm = protobuf(require('../../src/types/message/message.proto'))
 const CID = require('cids')
 
+const loadFixture = require('aegir/fixtures')
+const testDataPath = '../test-data/serialized-from-go'
+const rawMessageFullWantlist = loadFixture(__dirname, testDataPath + '/bitswap110-message-full-wantlist')
+const rawMessageOneBlock = loadFixture(__dirname, testDataPath + '/bitswap110-message-one-block')
+
 const BitswapMessage = require('../../src/types/message')
 
 describe('BitswapMessage', () => {
@@ -39,47 +44,100 @@ describe('BitswapMessage', () => {
     })
   })
 
-  it('.serializeToBitswap100', (done) => {
+  it('.serializeToBitswap100', () => {
     const block = blocks[1]
     const cid = cids[1]
-    const m = new BitswapMessage(true)
-    m.addBlock(cid, block)
-    expect(pbm.Message.decode(m.serializeToBitswap100()).blocks)
-      .to.be.eql([block.data])
-    done()
+    const msg = new BitswapMessage(true)
+    msg.addBlock(cid, block)
+    const serialized = msg.serializeToBitswap100()
+    expect(pbm.Message.decode(serialized).blocks).to.eql([block.data])
   })
 
-  it('.deserialize', (done) => {
-    const cid = cids[0]
+  it('.serializeToBitswap110', () => {
+    const block = blocks[1]
+    const cid = cids[1]
+    const msg = new BitswapMessage(true)
+    msg.addBlock(cid, block)
+    const serialized = msg.serializeToBitswap110()
+    expect(pbm.Message.decode(serialized).blocks).to.eql([block.data])
+  })
+
+  it('.deserialize a Bitswap100 Message', (done) => {
+    const cid0 = cids[0]
+    const cid1 = cids[1]
+    const cid2 = cids[2]
+
+    const b1 = blocks[1]
+    const b2 = blocks[2]
+
     const raw = pbm.Message.encode({
       wantlist: {
         entries: [{
-          block: cid.buffer,
+          block: cid0.buffer,
           cancel: false
         }],
         full: true
       },
       blocks: [
-        new Buffer('hello'),
-        new Buffer('world')
+        b1.data,
+        b2.data
       ]
     })
 
-    BitswapMessage.deserialize(raw, (err, protoMessage) => {
+    BitswapMessage.deserialize(raw, (err, msg) => {
       expect(err).to.not.exist
-      expect(protoMessage.full).to.equal(true)
-      expect(Array.from(protoMessage.wantlist))
-        .to.be.eql([[
-          cid.toBaseEncodedString(),
-          new BitswapMessage.Entry(cid, 0, false)
+      expect(msg.full).to.equal(true)
+      expect(Array.from(msg.wantlist))
+        .to.eql([[
+          cid0.toBaseEncodedString(),
+          new BitswapMessage.Entry(cid0, 0, false)
         ]])
 
-      const b1 = blocks[1]
-      const b2 = blocks[2]
-      const cid1 = cids[1]
-      const cid2 = cids[2]
+      expect(Array.from(msg.blocks).map((b) => [b[0], b[1].data]))
+        .to.eql([
+          [cid1.toBaseEncodedString(), b1.data],
+          [cid2.toBaseEncodedString(), b2.data]
+        ])
 
-      expect(Array.from(protoMessage.blocks).map((b) => [b[0], b[1].data]))
+      done()
+    })
+  })
+
+  it('.deserialize a Bitswap110 Message', (done) => {
+    const cid0 = cids[0]
+    const cid1 = cids[1]
+    const cid2 = cids[2]
+
+    const b1 = blocks[1]
+    const b2 = blocks[2]
+
+    const raw = pbm.Message.encode({
+      wantlist: {
+        entries: [{
+          block: cid0.buffer,
+          cancel: false
+        }],
+        full: true
+      },
+      payload: [{
+        data: b1.data,
+        prefix: cid1.prefix
+      }, {
+        data: b2.data,
+        prefix: cid2.prefix
+      }]
+    })
+
+    BitswapMessage.deserialize(raw, (err, msg) => {
+      expect(err).to.not.exist
+      expect(msg.full).to.equal(true)
+      expect(Array.from(msg.wantlist))
+        .to.eql([[
+          cid0.toBaseEncodedString(),
+          new BitswapMessage.Entry(cid0, 0, false)
+        ]])
+
+      expect(Array.from(msg.blocks).map((b) => [b[0], b[1].data]))
         .to.eql([
           [cid1.toBaseEncodedString(), b1.data],
           [cid2.toBaseEncodedString(), b2.data]
@@ -193,6 +251,24 @@ describe('BitswapMessage', () => {
       })
     })
 
-    it.skip('bitswap 1.1.0 message', (done) => {})
+    describe('bitswap 1.1.0 message', () => {
+      it('full wantlist message', (done) => {
+        BitswapMessage.deserialize(rawMessageFullWantlist, (err, message) => {
+          expect(err).to.not.exist
+          // TODO
+          //   check the deserialised message
+          done()
+        })
+      })
+
+      it('one block message', (done) => {
+        BitswapMessage.deserialize(rawMessageOneBlock, (err, message) => {
+          expect(err).to.not.exist
+          // TODO
+          //   check the deserialised message
+          done()
+        })
+      })
+    })
   })
 })