Merge pull request #39 from cabal-club/simpler-swarm

PROTOCOL BREAKING: use discovery-swarm for sending peer key

README.md

@@ -1,6 +1,6 @@
 # cabal-core
 
-Core database, replication, and chat APIs for cabal.
+Core database, replication, swarming, and chat APIs for cabal.
 
 ## Usage
 
@@ -20,7 +20,7 @@ If this is a new database, `key` can be omitted and will be generated.
 
 ### cabal.getLocalKey(cb)
 
-Returns the local user's key (as a string).
+Returns the local user's key (as a hex string).
 
 ### var ds = cabal.replicate()
 
@@ -55,6 +55,14 @@ Calls `fn` with every new message that arrives in `channel`.
 
 ### Network
 
+> var swarm = require('cabal-core/swarm')
+
+#### cabal.swarm(cb)
+
+Joins the P2P swarm for a cabal. This seeks out peers who are also part of this cabal by various means (internet, local network), connects to them, and replicates cabal messages between them.
+
+The returned object is an instance of [discovery-swarm](https://github.com/mafintosh/discovery-swarm).
+
 #### cabal.on('peer-added', function (key) {})
 
 Emitted when you connect to a peer. `key` is a hex string of their public key.
@@ -99,16 +107,6 @@ documented types include
 }
 ```
 
-### swarm
-
-> var swarm = require('cabal-core/swarm')
-
-#### swarm(cabal)
-
-Join the P2P swarm for a cabal, start connecting to peers and replicating messages.
-
-Returns a [discovery-swarm](https://github.com/mafintosh/discovery-swarm).
-
 ## License
 
 AGPLv3

index.js

@@ -8,6 +8,7 @@ var createChannelView = require('./views/channels')
 var createMessagesView = require('./views/messages')
 var createTopicsView = require('./views/topics')
 var createUsersView = require('./views/users')
+var swarm = require('./swarm')
 
 var DATABASE_VERSION = 1
 
@@ -105,7 +106,7 @@ Cabal.prototype.publish = function (message, opts, cb) {
   if (!opts) opts = {}
 
   this.feed(function (feed) {
-    message.timestamp = timestamp()
+    message.timestamp = message.timestamp || timestamp()
     feed.append(message, function (err) {
       cb(err, err ? null : message)
     })
@@ -154,14 +155,14 @@ Cabal.prototype.getLocalKey = function (cb) {
   })
 }
 
-/**
- * Replication stream for the mesh.
- */
-Cabal.prototype.replicate = function () {
-  return this.db.replicate({
-    live: true,
-    maxFeeds: this.maxFeeds
-  })
+Cabal.prototype.swarm = function (cb) {
+  swarm(this, cb)
+}
+
+Cabal.prototype.replicate = function (opts) {
+  opts = opts || {}
+  opts = Object.assign({}, {live:true}, opts)
+  return this.db.replicate(opts)
 }
 
 Cabal.prototype._addConnection = function (key) {

package.json

@@ -45,12 +45,13 @@
     "read-only-stream": "^2.0.0",
     "strftime": "^0.10.0",
     "through2": "^2.0.3",
-    "thunky": "^1.0.2",
+    "thunky": "^1.0.3",
     "xtend": "^4.0.1"
   },
   "devDependencies": {
     "collect-stream": "^1.2.1",
     "documentation": "^6.3.2",
+    "pump": "^3.0.0",
     "random-access-memory": "^3.0.0",
     "standard": "^11.0.1",
     "tape": "^4.9.1"

swarm.js

@@ -1,52 +1,29 @@
+var pump = require('pump')
 var discovery = require('discovery-swarm')
 var swarmDefaults = require('dat-swarm-defaults')
 
-module.exports = function (cabal) {
-  var swarm = discovery(swarmDefaults())
-  swarm.join(cabal.key.toString('hex'))
-  swarm.on('connection', function (conn, info) {
-    var remoteKey
-    var ended = false
+module.exports = function (cabal, cb) {
+  cb = cb || function () {}
 
-    cabal.getLocalKey(function (err, key) {
-      if (key) {
-        // send local key to remote
-        conn.write(new Buffer(key, 'hex'))
+  cabal.getLocalKey(function (err, key) {
+    if (err) return cb(err)
 
-        // read remote key from remote
-        conn.once('readable', onReadable)
+    var swarm = discovery(Object.assign({}, swarmDefaults(), { id: Buffer.from(key, 'hex') }))
+    swarm.join(cabal.key.toString('hex'))
+    swarm.on('connection', function (conn, info) {
+      var remoteKey = info.id.toString('hex')
+      conn.once('error', function () { if (remoteKey) cabal._removeConnection(remoteKey) })
+      conn.once('end',   function () { if (remoteKey) cabal._removeConnection(remoteKey) })
 
-        conn.once('end', function () {
-          ended = true
-        })
-
-        function onReadable () {
-          if (ended) return
-          var rkey = conn.read(32)
-          if (!rkey) {
-            conn.once('readable', onReadable)
-            return
-          }
+      var r = cabal.replicate()
+      pump(conn, r, conn, function (err) {
+        // TODO: report somehow
+      })
 
-          remoteKey = rkey.toString('hex')
-          cabal._addConnection(remoteKey)
-          replicate()
-        }
-      } else {
-        throw new Error('UNEXPECTED STATE: no local key!')
-      }
+      cabal._addConnection(remoteKey)
     })
 
-    function replicate () {
-      var r = cabal.replicate()
-      conn.pipe(r).pipe(conn)
-      r.on('error', noop)
-    }
-
-    conn.once('error', function () { if (remoteKey) cabal._removeConnection(remoteKey) })
-    conn.once('end',   function () { if (remoteKey) cabal._removeConnection(remoteKey) })
+    cb(null, swarm)
   })
-  return swarm
 }
 
-function noop () {}

test/test.js

@@ -2,6 +2,7 @@ var collect = require('collect-stream')
 var Cabal = require('..')
 var test = require('tape')
 var ram = require('random-access-memory')
+var pump = require('pump')
 
 test('create a cabal + channel', function (t) {
   var cabal = Cabal(ram)
@@ -130,3 +131,139 @@ test('listening for live messages', function (t) {
     })
   })
 })
+
+test('local replication', function (t) {
+  t.plan(15)
+
+  function create (id, cb) {
+    var cabal = Cabal(ram)
+    cabal.db.ready(function () {
+      var msg = {
+        type: 'chat/text',
+        content: {
+          text: 'hello from ' + id,
+          channel: 'general',
+          timestamp: Number(id) * 1000
+        }
+      }
+      cabal.getLocalKey(function (err, key) {
+        if (err) return cb(err)
+        cabal.key = key
+        cabal.publish(msg, function (err) {
+          if (err) cb(err)
+          else cb(null, cabal)
+        })
+      })
+    })
+  }
+
+  create(1, function (err, c1) {
+    t.error(err)
+    create(2, function (err, c2) {
+      t.error(err)
+      sync(c1, c2, function (err) {
+        t.error(err, 'sync ok')
+
+        function check (cabal) {
+          var r = cabal.messages.read('general')
+          collect(r, function (err, data) {
+            t.error(err)
+            t.same(data.length, 2, '2 messages')
+            t.same(data[0].key, c2.key)
+            t.same(data[0].seq, 0)
+            t.same(data[1].key, c1.key)
+            t.same(data[1].seq, 0)
+          })
+        }
+
+        check(c1)
+        check(c2)
+      })
+    })
+  })
+})
+
+test('swarm network replication', function (t) {
+  t.plan(15)
+
+  function create (id, cb) {
+    var cabal = Cabal(ram, 'fake')
+    cabal.db.ready(function () {
+      var msg = {
+        type: 'chat/text',
+        content: {
+          text: 'hello from ' + id,
+          channel: 'general',
+          timestamp: Number(id) * 1000
+        }
+      }
+      cabal.getLocalKey(function (err, key) {
+        if (err) return cb(err)
+        cabal._localkey = key
+        cabal.publish(msg, function (err) {
+          if (err) cb(err)
+          else cb(null, cabal)
+        })
+      })
+    })
+  }
+
+  create(1, function (err, c1) {
+    t.error(err)
+    create(2, function (err, c2) {
+      t.error(err)
+      syncNetwork(c1, c2, function (err) {
+        t.error(err, 'sync ok')
+
+        function check (cabal) {
+          var r = cabal.messages.read('general')
+          collect(r, function (err, data) {
+            t.error(err)
+            t.same(data.length, 2, '2 messages')
+            t.same(data[0].key, c2._localkey)
+            t.same(data[0].seq, 0)
+            t.same(data[1].key, c1._localkey)
+            t.same(data[1].seq, 0)
+          })
+        }
+
+        check(c1)
+        check(c2)
+      })
+    })
+  })
+})
+
+function sync (a, b, cb) {
+  var r = a.replicate({live:false})
+  pump(r, b.replicate({live:false}), r, cb)
+}
+
+function syncNetwork (a, b, cb) {
+  var pending = 2
+
+  a.swarm(function (err, swarm1) {
+    if (err) return cb(err)
+    b.swarm(function (err, swarm2) {
+      if (err) return cb(err)
+
+      function end () {
+        if (!--pending) {
+          swarm1.destroy(function () {
+            swarm2.destroy(cb)
+          })
+        }
+      }
+
+      a.on('peer-added', function (key) {
+        console.log('a-add', key)
+        setTimeout(end, 2000)
+      })
+
+      b.on('peer-added', function (key) {
+        console.log('b-add', key)
+        setTimeout(end, 2000)
+      })
+    })
+  })
+}