Merge pull request #6 from ipfs/fix

fix double get

src/decision/engine.js

@@ -45,7 +45,7 @@ module.exports = class Engine {
         doIt(() => {
           this._timer = null
         })
-      }, 200)
+      }, 100)
     }
 
     const doIt = (cb) => {
@@ -115,6 +115,22 @@ module.exports = class Engine {
       })
   }
 
+  receivedBlock (block) {
+    this._processBlock(block)
+    this._outbox()
+  }
+
+  _processBlock (block) {
+    // Check all connected peers if they want the block we received
+    for (let l of this.ledgerMap.values()) {
+      const entry = l.wantlistContains(block.key)
+
+      if (entry) {
+        this.peerRequestQueue.push(entry, l.partner)
+      }
+    }
+  }
+
   _processWantlist (ledger, peerId, entry, cb) {
     if (entry.cancel) {
       log('cancel %s', entry.key)
@@ -142,14 +158,7 @@ module.exports = class Engine {
       log('got block %s %s bytes', block.key, block.data.length)
       ledger.receivedBytes(block.data.length)
 
-      // Check all connected peers if they want the block we received
-      for (let l of this.ledgerMap.values()) {
-        const entry = l.wantlistContains(block.key)
-
-        if (entry) {
-          this.peerRequestQueue.push(entry, ledger.partner)
-        }
-      }
+      this.receivedBlock(block)
     }
   }
 

src/index.js

@@ -190,12 +190,15 @@ module.exports = class Bitwap {
 
   // announces the existance of a block to this service
   hasBlock (block, cb) {
+    cb = cb || (() => {})
+
     this._tryPutBlock(block, 4, (err) => {
       if (err) {
-        log.error('Error writing block to datastor: %s', err.message)
+        log.error('Error writing block to datastore: %s', err.message)
         return cb(err)
       }
       this.notifications.emit(`block:${block.key.toString('hex')}`, block)
+      this.engine.receivedBlock(block)
       cb()
     })
   }

test/index-test.js

@@ -11,6 +11,11 @@ const Message = require('../src/message')
 const Bitswap = require('../src')
 
 const mockNetwork = require('./utils').mockNetwork
+const applyNetwork = (bs, n) => {
+  bs.network = n
+  bs.wm.network = n
+  bs.engine.network = n
+}
 
 module.exports = (repo) => {
   describe('bitswap', () => {
@@ -189,9 +194,7 @@ module.exports = (repo) => {
         }
       }
       bs1 = new Bitswap(me, libp2p, store)
-      bs1.network = n1
-      bs1.wm.network = n1
-      bs1.engine.network = n1
+      applyNetwork(bs1, n1)
 
       let store2
 
@@ -203,9 +206,7 @@ module.exports = (repo) => {
         },
         (val, cb) => {
           bs2 = new Bitswap(other, libp2p, store2)
-          bs2.network = n2
-          bs2.wm.network = n2
-          bs2.engine.network = n2
+          applyNetwork(bs2, n2)
           bs1._onPeerConnected(other)
           bs2._onPeerConnected(me)
           bs1.getBlock(block.key, cb)
@@ -236,6 +237,74 @@ module.exports = (repo) => {
         bs.hasBlock(block, () => {})
       }, 200)
     })
+
+    it('block is sent after local add', (done) => {
+      const me = PeerId.create({bit: 64})
+      const other = PeerId.create({bit: 64})
+      const libp2p = {}
+      const block = new Block('hello world local add')
+      let bs1
+      let bs2
+      let n1
+      let n2
+
+      n1 = {
+        connectTo (id, cb) {
+          let err
+          if (id.toHexString() !== other.toHexString()) {
+            err = new Error('unkown peer')
+          }
+          async.setImmediate(() => cb(err))
+        },
+        sendMessage (id, msg, cb) {
+          if (id.toHexString() === other.toHexString()) {
+            bs2._receiveMessage(me, msg, cb)
+          } else {
+            async.setImmediate(() => cb(new Error('unkown peer')))
+          }
+        }
+      }
+      n2 = {
+        connectTo (id, cb) {
+          let err
+          if (id.toHexString() !== me.toHexString()) {
+            err = new Error('unkown peer')
+          }
+          async.setImmediate(() => cb(err))
+        },
+        sendMessage (id, msg, cb) {
+          if (id.toHexString() === me.toHexString()) {
+            bs1._receiveMessage(other, msg, cb)
+          } else {
+            async.setImmediate(() => cb(new Error('unkown peer')))
+          }
+        }
+      }
+      bs1 = new Bitswap(me, libp2p, store)
+      applyNetwork(bs1, n1)
+
+      let store2
+
+      async.waterfall([
+        (cb) => repo.create('world', cb),
+        (repo, cb) => {
+          store2 = repo.datastore
+          bs2 = new Bitswap(other, libp2p, store2)
+          applyNetwork(bs2, n2)
+          bs1._onPeerConnected(other)
+          bs2._onPeerConnected(me)
+          bs1.getBlock(block.key, cb)
+
+          setTimeout(() => {
+            bs2.hasBlock(block)
+          }, 1000)
+        },
+        (res, cb) => {
+          expect(res).to.be.eql(res)
+          cb()
+        }
+      ], done)
+    })
   })
 
   describe('stat', () => {