feature: ability to override default bitswap max message size

src/decision-engine/index.js

@@ -21,7 +21,7 @@ const logger = require('../utils').logger
 const MAX_MESSAGE_SIZE = 512 * 1024
 
 class DecisionEngine {
-  constructor (peerId, blockstore, network) {
+  constructor (peerId, blockstore, network, options) {
     this._log = logger(peerId, 'engine')
     this.blockstore = blockstore
     this.network = network
@@ -33,6 +33,13 @@ class DecisionEngine {
     // List of tasks to be processed
     this._tasks = []
 
+    options = options || {}
+    this._maxMessageSize = options.maxMessageSize || MAX_MESSAGE_SIZE
+    // override Max size to 1kb bytes by minimum.
+    if (this._maxMessageSize < 1024) {
+      this._maxMessageSize = 1024
+    }
+
     this._outbox = debounce(this._processTasks.bind(this), 100)
   }
 
@@ -42,7 +49,7 @@ class DecisionEngine {
       return acc + b.data.byteLength
     }, 0)
 
-    if (total < MAX_MESSAGE_SIZE) {
+    if (total < this._maxMessageSize) {
       return this._sendSafeBlocks(peer, blocks, cb)
     }
 
@@ -55,7 +62,7 @@ class DecisionEngine {
       batch.push(b)
       size += b.data.byteLength
 
-      if (size >= MAX_MESSAGE_SIZE ||
+      if (size >= this._maxMessageSize ||
           // need to ensure the last remaining items get sent
           outstanding === 0) {
         const nextBatch = batch.slice()

src/index.js

@@ -21,17 +21,17 @@ const logger = require('./utils').logger
  * @param {Blockstore} blockstore
  */
 class Bitswap {
-  constructor (libp2p, blockstore) {
+  constructor (libp2p, blockstore, options) {
     this._libp2p = libp2p
     this._log = logger(this.peerInfo.id)
-
+    options = options || {}
     // the network delivers messages
     this.network = new Network(libp2p, this)
 
     // local database
     this.blockstore = blockstore
 
-    this.engine = new DecisionEngine(this.peerInfo.id, blockstore, this.network)
+    this.engine = new DecisionEngine(this.peerInfo.id, blockstore, this.network, options)
 
     // handle message sending
     this.wm = new WantManager(this.peerInfo.id, this.network)

test/bitswap.js

@@ -18,8 +18,15 @@ const createLibp2pNode = require('./utils/create-libp2p-node')
 const makeBlock = require('./utils/make-block')
 const orderedFinish = require('./utils/helpers').orderedFinish
 
+const MAX_MESSAGE_SIZE = 512 * 1024
+
 // Creates a repo + libp2pNode + Bitswap with or without DHT
-function createThing (dht, callback) {
+function createThing (dht, msgSize, callback) {
+  if (!callback) {
+    callback = msgSize
+    msgSize = MAX_MESSAGE_SIZE
+  }
+
   waterfall([
     (cb) => createTempRepo(cb),
     (repo, cb) => {
@@ -28,7 +35,7 @@ function createThing (dht, callback) {
       }, (err, node) => cb(err, repo, node))
     },
     (repo, libp2pNode, cb) => {
-      const bitswap = new Bitswap(libp2pNode, repo.blocks)
+      const bitswap = new Bitswap(libp2pNode, repo.blocks, {maxMessageSize: msgSize})
       bitswap.start((err) => cb(err, repo, libp2pNode, bitswap))
     }
   ], (err, repo, libp2pNode, bitswap) => {
@@ -49,12 +56,13 @@ describe('bitswap without DHT', function () {
 
   before((done) => {
     parallel([
-      (cb) => createThing(false, cb),
-      (cb) => createThing(false, cb),
-      (cb) => createThing(false, cb)
+      (cb) => createThing(false, MAX_MESSAGE_SIZE, cb),
+      (cb) => createThing(false, MAX_MESSAGE_SIZE, cb),
+      (cb) => createThing(false, MAX_MESSAGE_SIZE, cb),
+      (cb) => createThing(false, 32, cb)
     ], (err, results) => {
       expect(err).to.not.exist()
-      expect(results).to.have.length(3)
+      expect(results).to.have.length(4)
       nodes = results
       done()
     })
@@ -77,6 +85,13 @@ describe('bitswap without DHT', function () {
     ], done)
   })
 
+  it('connect 1 -> 3 && 2 -> 3', (done) => {
+    parallel([
+      (cb) => nodes[1].libp2pNode.dial(nodes[3].libp2pNode.peerInfo, cb),
+      (cb) => nodes[2].libp2pNode.dial(nodes[3].libp2pNode.peerInfo, cb)
+    ], done)
+  })
+
   it('put a block in 2, fail to get it in 0', (done) => {
     const finish = orderedFinish(2, done)
 
@@ -104,12 +119,13 @@ describe('bitswap with DHT', () => {
 
   before((done) => {
     parallel([
-      (cb) => createThing(true, cb),
-      (cb) => createThing(true, cb),
-      (cb) => createThing(true, cb)
+      (cb) => createThing(true, MAX_MESSAGE_SIZE, cb),
+      (cb) => createThing(true, MAX_MESSAGE_SIZE, cb),
+      (cb) => createThing(true, MAX_MESSAGE_SIZE, cb),
+      (cb) => createThing(true, 32, cb)
     ], (err, results) => {
       expect(err).to.not.exist()
-      expect(results).to.have.length(3)
+      expect(results).to.have.length(4)
       nodes = results
       done()
     })
@@ -125,10 +141,11 @@ describe('bitswap with DHT', () => {
     }, done)
   })
 
-  it('connect 0 -> 1 && 1 -> 2', (done) => {
+  it('connect 0 -> 1 && 1 -> 2 && 2 -> 3', (done) => {
     parallel([
       (cb) => nodes[0].libp2pNode.dial(nodes[1].libp2pNode.peerInfo, cb),
-      (cb) => nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerInfo, cb)
+      (cb) => nodes[1].libp2pNode.dial(nodes[2].libp2pNode.peerInfo, cb),
+      (cb) => nodes[2].libp2pNode.dial(nodes[3].libp2pNode.peerInfo, cb)
     ], done)
   })
 
@@ -145,4 +162,32 @@ describe('bitswap with DHT', () => {
       })
     ], done)
   })
+
+  it('put a block in 2, get it in 3', (done) => {
+    waterfall([
+      (cb) => makeBlock(cb),
+      (block, cb) => nodes[2].bitswap.put(block, () => cb(null, block)),
+      (block, cb) => setTimeout(() => cb(null, block), 400),
+      (block, cb) => nodes[3].bitswap.get(block.cid, (err, blockRetrieved) => {
+        expect(err).to.not.exist()
+        expect(block.data).to.eql(blockRetrieved.data)
+        expect(block.cid).to.eql(blockRetrieved.cid)
+        cb()
+      })
+    ], done)
+  })
+
+  it('put a block in 3, get it in 2', (done) => {
+    waterfall([
+      (cb) => makeBlock(cb),
+      (block, cb) => nodes[3].bitswap.put(block, () => cb(null, block)),
+      (block, cb) => setTimeout(() => cb(null, block), 400),
+      (block, cb) => nodes[2].bitswap.get(block.cid, (err, blockRetrieved) => {
+        expect(err).to.not.exist()
+        expect(block.data).to.eql(blockRetrieved.data)
+        expect(block.cid).to.eql(blockRetrieved.cid)
+        cb()
+      })
+    ], done)
+  })
 })

test/types/message.spec.js

@@ -266,6 +266,7 @@ describe('BitswapMessage', () => {
           expect(err).to.not.exist()
           // TODO
           //   check the deserialised message
+          expect(message.serializeToBitswap110()).to.eql(rawMessageFullWantlist)
           done()
         })
       })
@@ -275,6 +276,7 @@ describe('BitswapMessage', () => {
           expect(err).to.not.exist()
           // TODO
           //   check the deserialised message
+          expect(message.serializeToBitswap110()).to.eql(rawMessageOneBlock)
           done()
         })
       })