Merge pull request #8 from datproject/add/read-manifest-updates

add live update version of .readManifest

README.md

@@ -48,6 +48,10 @@ Close and remove a `dat` archive.
 Read the `dat.json` file from the `dat` archive. This method is expected to be
 deprecated once `dat` archives provide a built-in method to return archives.
 
+### updates = multidat.readManifest(dat)
+Subscribe to updates to the `dat.json` file and emits `"manifest"` events.
+Call `updates.stop()` to stop listening.
+
 ## Why?
 This package exists to manage multiple `dat` archives in different directories.
 The [dat-node][dat-node] package is mostly stateless; all state is persisted

index.js

@@ -1,3 +1,4 @@
+var EventEmitter = require('events').EventEmitter
 var multidrive = require('multidrive')
 var explain = require('explain-error')
 var parse = require('fast-json-parse')
@@ -55,6 +56,12 @@ function Multidat (db, opts, cb) {
   }
 
   function closeArchive (dat, done) {
+    if (dat._listStreams) {
+      dat._listStreams.forEach(function (listStream) {
+        listStream.destroy()
+      })
+      dat._listStreams = []
+    }
     dat.close(function (err) {
       if (err) return done(explain(err, 'multidat.closeArchive: error closing dat archive'))
       dat.db.close(done)
@@ -63,21 +70,40 @@ function Multidat (db, opts, cb) {
 }
 
 function readManifest (dat, done) {
+  var updates = new EventEmitter()
+
+  if (done) {
+    updates.on('error', done)
+    updates.once('manifest', function (manifest) {
+      updates.stop()
+      done(null, manifest)
+    })
+  }
+
   var listStream = dat.archive.list({ live: true })
+  dat._listStreams = dat._listStreams || []
+  dat._listStreams.push(listStream)
   listStream.on('data', function (entry) {
     if (entry.name !== 'dat.json') return
 
     var rs = dat.archive.createFileReadStream('dat.json')
     var ws = concat(sink)
     pump(rs, ws, function (err) {
-      if (err) return done(explain(err, 'multidat.readManifest: error piping data'))
+      if (err) return updates.emit('error', explain(err, 'multidat.readManifest: error piping data'))
     })
   })
 
   function sink (data) {
-    listStream.destroy()
     var res = parse(data)
-    if (res.err) return done(explain(res.err, "multidat.readManifest: couldn't parse dat.json file"))
-    done(null, res.value)
+    if (res.err) return updates.emit('error', explain(res.err, "multidat.readManifest: couldn't parse dat.json file"))
+    updates.emit('manifest', res.value)
+  }
+
+  updates.stop = function () {
+    var idx = dat._listStreams.indexOf(listStream)
+    if (idx === -1) return
+    listStream.destroy()
+    dat._listStreams.splice(idx, 1)
   }
+  return updates
 }

test.js

@@ -151,5 +151,83 @@ tape('multidat = Multidat()', function (t) {
         })
       })
     })
+
+    t.test('should subscribe to updates', function (t) {
+      t.plan(5)
+      var driveDb = memdb()
+      var drive = hyperdrive(driveDb)
+      var archive = drive.createArchive()
+      var ws = archive.createFileWriteStream('dat.json')
+      var swarm = hyperdiscovery(archive)
+      ws.end(JSON.stringify({ name: 'hello-planet' }))
+
+      var db = toilet({})
+      Multidat(db, opts, function (err, multidat) {
+        t.ifError(err, 'no error')
+
+        var location = path.join('/tmp', String(Date.now()))
+        mkdirp.sync(location)
+
+        multidat.create(location, { key: archive.key }, function (err, dat) {
+          t.ifError(err, 'no error')
+
+          if (!worker) dat.joinNetwork()
+          var updates = multidat.readManifest(dat)
+          updates.on('error', function (err) {
+            t.ifError(err, 'no err')
+          })
+          updates.on('manifest', function (manifest) {
+            t.equal(typeof manifest, 'object', 'right type')
+            t.equal(manifest.name, 'hello-planet', 'right value')
+            updates.stop()
+            dat.close(function () {
+              swarm.close(function () {
+                t.pass('done closing')
+                rimraf.sync(location)
+              })
+            })
+          })
+        })
+      })
+    })
+
+    t.test('subscribtions should be cleaned up', function (t) {
+      t.plan(6)
+      var driveDb = memdb()
+      var drive = hyperdrive(driveDb)
+      var archive = drive.createArchive()
+      var ws = archive.createFileWriteStream('dat.json')
+      var swarm = hyperdiscovery(archive)
+      ws.end(JSON.stringify({ name: 'hello-planet' }))
+
+      var db = toilet({})
+      Multidat(db, opts, function (err, multidat) {
+        t.ifError(err, 'no error')
+
+        var location = path.join('/tmp', String(Date.now()))
+        mkdirp.sync(location)
+
+        multidat.create(location, { key: archive.key }, function (err, dat) {
+          t.ifError(err, 'no error')
+
+          if (!worker) dat.joinNetwork()
+          var updates = multidat.readManifest(dat)
+          updates.on('error', function (err) {
+            t.ifError(err, 'no err')
+          })
+          updates.on('manifest', function (manifest) {
+            t.equal(typeof manifest, 'object', 'right type')
+            t.equal(manifest.name, 'hello-planet', 'right value')
+            multidat.close(dat.key, function (err) {
+              t.ifError(err, 'no error')
+              swarm.close(function () {
+                t.pass('done closing')
+                rimraf.sync(location)
+              })
+            })
+          })
+        })
+      })
+    })
   })
 })