rewriting blobs plugin and replication

.travis.yml

@@ -0,0 +1,4 @@
+language: node_js
+node_js:
+  - 0.6
+  - 0.8

LICENSE

@@ -0,0 +1,22 @@
+Copyright (c) 2016 Dominic Tarr
+
+Permission is hereby granted, free of charge, 
+to any person obtaining a copy of this software and 
+associated documentation files (the "Software"), to 
+deal in the Software without restriction, including 
+without limitation the rights to use, copy, modify, 
+merge, publish, distribute, sublicense, and/or sell 
+copies of the Software, and to permit persons to whom 
+the Software is furnished to do so, 
+subject to the following conditions:
+
+The above copyright notice and this permission notice 
+shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 
+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. 
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR 
+ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, 
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE 
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,23 @@
+# ssb-blobs
+
+current protocol:
+
+client calls `has([id,...], cb)` and then
+server calls back array `[boolean: has,...]`
+
+if the server reports they have a given blob,
+then the client requests that blob.
+
+## want-stream
+
+Client creates want stream.
+
+Client & server emit their wants {hash: hops} tuples.
+if a peer has a hash, they respond: {hash: size}
+
+if that size is under the max, request the blob from that peer.
+delete the blob from the want list.
+
+## License
+
+MIT

index.js

@@ -0,0 +1,174 @@
+
+function isEmpty (o) {
+  for(var k in o) return false
+  return true
+}
+
+function isInteger (i) {
+  return Number.isInteger(i)
+}
+
+var Notify = require('pull-notify')
+var pull = require('pull-stream')
+var isBlobId = require('ssb-ref').isBlob
+
+var MB = 1024*1024
+var MAX_SIZE = 5*MB
+
+function noop () {}
+
+module.exports = function (blobs) {
+
+  var notify = Notify()
+  var changes = Notify()
+
+  var want = {}, waiting = {}
+
+  var send = {}, timer
+
+  function queue (hash, hops) {
+    console.log('Queue', hash, hops, send)
+    if(hops < 0)
+      want[hash] = hops
+    else
+      delete want[hash]
+
+    send[hash] = hops
+  //  clearTimeout(timer)
+//    timer = setTimeout(function () {
+      var _send = send
+      send = {}
+      notify(_send)
+    //}, 10)
+  }
+
+  var peers = {}
+
+  function add(id, cb) {
+    cb = cb || noop
+    return blobs.add(id, function (err, id) {
+      if(err) cb(err)
+      else cb(null, changes(id))
+    })
+  }
+
+  function get (peer, id) {
+    pull(
+      peers[peer].blobs.get(id),
+      add(id, function (err, h) {
+        console.log('GOT', h)
+      })
+    )
+  }
+
+  function wants (peer, id, hops) {
+    if(!want[id] || want[id] < hops) {
+      want[id] = hops
+      queue(id, hops)
+    }
+  }
+
+  pull(
+    changes.listen(),
+    pull.drain(function (id) {
+      blobs.size(id, function (err, size) {
+        if(size) queue(id, size)
+      })
+      delete want[id]
+      if(waiting[id])
+        while(waiting[id].length)
+          waiting[id].shift()(null, true)
+    })
+  )
+
+  function has(peer, id, size) {
+    console.log('HAS-GET?', want, peer, size)
+    if(want[id] && size < MAX_SIZE) get(peer, id)
+  }
+
+  function process (data, peer, cb) {
+    var n = 0, res = {}
+    for(var id in data) {
+      if(isBlobId(id) && isInteger(data[id])) {
+        if(data[id] <= 0) { //interpret as "WANT"
+          n++
+          //check whether we already *HAVE* this file.
+          //respond with it's size, if we do.
+          blobs.size(id, function (err, size) {
+            if(size) res[id] = size
+            else wants(peer, id, data[id] - 1)
+            next()
+          })
+        }
+        else if(data[id] > 0) { //interpret as "HAS"
+          has(peer, id, data[id])
+        }
+      }
+    }
+
+    function next () {
+      if(--n) return
+      cb(null, res)
+    }
+  }
+
+  return {
+    get: blobs.get,
+    add: add,
+    changes: function () {
+      return changes.listen()
+    },
+    want: function (hash, cb) {
+      console.log('WANT', hash)
+      if(waiting[hash])
+        waiting[hash].push(cb)
+      else {
+        waiting[hash] = [cb]
+        blobs.has(hash, function (err, has) {
+          if(has) {
+            while(wating[hash].length)
+              wating[hash].shift()(null, true)
+            delete waiting[hash]
+          }
+          else
+            queue(hash, -1)
+        })
+      }
+    },
+    createWantStream: function () {
+      var source = notify.listen()
+      var peer = this
+      peers[peer.id] = this
+      //queue current want list...
+      return {
+        source: source,
+        sink: pull.drain(function (data) {
+          //respond with list of blobs you already have,
+          process(data, peer.id, function (err, has_data) {
+            //(if you have any)
+            if(!isEmpty(has_data)) source.push(has_data)
+          })
+        }, function (_) {
+          if(peers[peer.id] == peer)
+            delete peers[peer.id]
+        })
+      }
+    }
+  }
+}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+

package.json

@@ -0,0 +1,22 @@
+{
+  "name": "ssb-blobs",
+  "description": "",
+  "version": "0.0.0",
+  "homepage": "https://github.com/dominictarr/ssb-blobs",
+  "repository": {
+    "type": "git",
+    "url": "git://github.com/dominictarr/ssb-blobs.git"
+  },
+  "dependencies": {
+    "pull-notify": "0.0.2",
+    "ssb-ref": "^2.3.0"
+  },
+  "devDependencies": {
+    "tape": "^4.5.1"
+  },
+  "scripts": {
+    "test": "set -e; for t in test/*.js; do node $t; done"
+  },
+  "author": "Dominic Tarr <dominic.tarr@gmail.com> (http://dominictarr.com)",
+  "license": "MIT"
+}

test/index.js

@@ -0,0 +1,125 @@
+var Notify = require('pull-notify')
+var crypto = require('crypto')
+var tape = require('tape')
+var Blobs = require('../')
+var pull = require('pull-stream')
+
+var MockBlobStore = require('./mock')
+
+function createHash () {
+  var args = [].slice.call(arguments)
+  var o = {}
+  for(var i = 0; i < args.length; i+=2) o[args[i]] = args[i+1]
+  return o
+}
+
+//create random value that looks like a blob id.
+function rand() {
+  return '&'+crypto.randomBytes(32).toString('base64')+'.sha256'
+}
+
+function Fake(string, length) {
+  var b = new Buffer(length)
+  var n = Buffer.byteLength(string)
+  for(var i = 0; i < length; i += n)
+    b.write(string, i)
+  return b
+}
+
+function hash(buf) {
+  buf = 'string' == typeof buf ? new Buffer(buf) : buf
+  return '&'+crypto.createHash('sha256')
+            .update(buf).digest('base64')+'.sha256'
+}
+
+
+tape('simple', function (t) {
+  var blobs = Blobs(MockBlobStore())
+
+  var b = Fake('hello', 256), h = hash(b)
+  pull(pull.once(b), blobs.add())
+
+  var req = {}
+  req[h] = 0
+  var res = {}
+  res[h] = 256
+
+  pull(
+    pull.once(req),
+    blobs.createWantStream(),
+    pull.find(null, function (err, _res) {
+      if(err) throw err
+      //requests 
+      t.deepEqual(_res, res)
+      t.end()
+    })
+  )
+})
+
+tape('want', function (t) {
+  var blobs = Blobs(MockBlobStore())
+  var h = hash(Fake('foobar', 64))
+  var res = {}
+  res[h] = -1
+
+  pull(
+    blobs.createWantStream(),
+    pull.find(null, function (err, _res) {
+      if(err) throw err
+      //requests 
+      t.deepEqual(_res, res)
+      t.end()
+    })
+  )
+
+  blobs.want(h)
+})
+
+function log (name) {
+  return pull.through(function (e) {
+    console.log(name, e)
+  })
+}
+
+function peers (nameA, a, nameB, b) {
+  var as = a.createWantStream.call({id: nameB, blobs: b})
+  var bs = b.createWantStream.call({id: nameA, blobs: a})
+  var na = nameA[0].toUpperCase(), nb = nameB[0].toUpperCase()
+  pull(as, log(na+nb), bs, log(nb+na), as)
+}
+
+tape('want - has', function (t) {
+  var alice = Blobs(MockBlobStore())
+  var bob   = Blobs(MockBlobStore())
+  var blob = Fake('foobar', 64)
+  var h = hash(blob)
+
+  peers('alice', alice, 'bob', bob)
+
+  alice.want(h, function (err) {
+    t.end()
+  })
+
+  pull(pull.once(blob), bob.add())
+})
+
+tape('want - want -has', function (t) {
+  var alice = Blobs(MockBlobStore())
+  var bob   = Blobs(MockBlobStore())
+  var carol = Blobs(MockBlobStore())
+
+  var blob = Fake('baz', 64)
+  var h = hash(blob)
+
+  peers('alice', alice, 'bob', bob)
+  peers('bob', bob, 'carol', carol)
+
+  alice.want(h, function (err) {
+    t.end()
+  })
+
+  pull(pull.once(blob), carol.add())
+})
+
+
+