Merge 9e7793b into 6ad08e7

API.md

@@ -25,7 +25,24 @@ Fetch a single block.
 - `keys: []Multihash`
 - `cb: Function`
 
-Fetch multiple blocks.
+Fetch multiple blocks. The `cb` is called with a result object of the form
+```js
+{
+  [key1]: {error: errorOrUndefined, block: blockOrUndefined},
+  [key2]: {error: errorOrUndefined, block: blockOrUndefined},
+  ...
+}
+```
+
+Where `key<i>` is the multihash of the block.
+
+### `unwantBlocks(keys)`
+
+- `keys: []Multihash`
+
+Cancel previously requested keys, forcefully. That means they are removed from the
+wantlist independent of how many other resources requested these keys. Callbacks
+attached to `getBlock` are errored with `Error('manual unwant: key')`.
 
 ### `cancelWants(keys)`
 

package.json

@@ -71,4 +71,4 @@
     "David Dias <daviddias.p@gmail.com>",
     "Friedel Ziegelmayer <dignifiedquire@gmail.com>"
   ]
-}
+}

src/decision/engine.js

@@ -55,7 +55,6 @@ module.exports = class Engine {
         if (!nextTask) return push(null, _.nil)
 
         this.datastore.get(nextTask.entry.key, (err, block) => {
-          log('fetched: %s', block.key.toString('hex'), block.data.toString())
           if (err || !block) {
             nextTask.done()
           } else {

src/index.js

@@ -138,12 +138,11 @@ module.exports = class Bitwap {
       cb(err, block)
     }
 
-    this.getBlocks([key], (err, res) => {
-      if (err) {
-        return done(err)
-      }
+    this.getBlocks([key], (results) => {
+      const err = results[key].error
+      const block = results[key].block
 
-      done(null, res[0])
+      done(err, block)
     })
   }
 
@@ -153,18 +152,51 @@ module.exports = class Bitwap {
   }
 
   getBlocks (keys, cb) {
-    const blocks = []
-    const finish = (block) => {
-      blocks.push(block)
-      log('finish: %s/%s', blocks.length, keys.length)
-      if (blocks.length === keys.length) {
-        cb(null, blocks)
+    const results = {}
+    const unwantListeners = {}
+    const blockListeners = {}
+    const unwantEvent = (key) => `unwant:${key.toString('hex')}`
+    const blockEvent = (key) => `block:${key.toString('hex')}`
+
+    const cleanupListeners = () => {
+      keys.forEach((key) => {
+        this.notifications.removeListener(unwantEvent(key), unwantListeners[key])
+        this.notifications.removeListener(blockEvent(key), blockListeners[key])
+      })
+    }
+
+    const addListeners = () => {
+      keys.forEach((key) => {
+        unwantListeners[key] = () => {
+          finish(key, new Error(`manual unwant: ${key.toString('hex')}`))
+        }
+
+        blockListeners[key] = (block) => {
+          finish(key, null, block)
+        }
+
+        this.notifications.once(unwantEvent(key), unwantListeners[key])
+        this.notifications.once(blockEvent(key), blockListeners[key])
+      })
+    }
+
+    const finish = (key, err, block) => {
+      results[key] = {
+        error: err,
+        block: block
+      }
+
+      if (Object.keys(results).length === keys.length) {
+        cleanupListeners()
+        cb(results)
       }
     }
 
+    addListeners()
+
     keys.forEach((key) => {
-      // Sanity check, we don't want to announce looking for blocks
-      // when we might have them ourselves
+      // We don't want to announce looking for blocks
+      // when we might have them ourselves.
       this.datastore.has(key, (err, exists) => {
         if (err) {
           log('error in datastore.has: ', err.message)
@@ -174,8 +206,8 @@ module.exports = class Bitwap {
         if (exists) {
           this.datastore.get(key, (err, res) => {
             if (!err && res) {
+              finish(key, null, res)
               this.wm.cancelWants([key])
-              finish(res)
               return
             }
 
@@ -185,14 +217,19 @@ module.exports = class Bitwap {
           })
         }
       })
-      this.notifications.once(`block:${key.toString('hex')}`, (block) => {
-        finish(block)
-      })
     })
 
     this.wm.wantBlocks(keys)
   }
 
+  // removes the given keys from the want list independent of any ref counts
+  unwantBlocks (keys) {
+    this.wm.unwantBlocks(keys)
+    keys.forEach((key) => {
+      this.notifications.emit(`unwant:${key.toString('hex')}`)
+    })
+  }
+
   // removes the given keys from the want list
   cancelWants (keys) {
     this.wm.cancelWants(keys)

src/wantlist/index.js

@@ -35,6 +35,12 @@ class Wantlist {
     this.set.delete(key.toString('hex'))
   }
 
+  removeForce (key) {
+    if (this.set.has(key)) {
+      this.set.delete(key)
+    }
+  }
+
   entries () {
     return this.set.entries()
   }

src/wantmanager/index.js

@@ -23,7 +23,7 @@ module.exports = class Wantmanager {
     return new MsgQueue(peerId, this.network)
   }
 
-  _addEntries (keys, cancel) {
+  _addEntries (keys, cancel, force) {
     let i = -1
     _(keys)
       .map((key) => {
@@ -33,7 +33,11 @@ module.exports = class Wantmanager {
       .tap((e) => {
         // add changes to our wantlist
         if (e.cancel) {
-          this.wl.remove(e.key)
+          if (force) {
+            this.wl.removeForce(e.key)
+          } else {
+            this.wl.remove(e.key)
+          }
         } else {
           this.wl.add(e.key, e.priority)
         }
@@ -90,6 +94,12 @@ module.exports = class Wantmanager {
     this._addEntries(keys, false)
   }
 
+  // remove blocks of all the given keys without respecting refcounts
+  unwantBlocks (keys) {
+    log('unwant blocks:', keys)
+    this._addEntries(keys, true, true)
+  }
+
   // cancel wanting all of the given keys
   cancelWants (keys) {
     log('cancel wants: ', keys)