Additional methods and bug fixes

Methods added/updated:
* version
* createDiffStream
* download (untested)
* stat returns Stat objects.

Also updated the session management, and stopped writing the .key file
by default

Author: andrewosh

index.js

@@ -223,7 +223,7 @@ function wrap (metadata, methods, opts) {
       method(call)
         .then(rsp => {
           log.debug(tag, 'request was successful')
-          if (cb) return cb(null, rsp)
+          if (cb) process.nextTick(cb, null, rsp)
         })
         .catch(err => {
           log.error({ ...tag, error: err.toString(), stack: err.stack }, 'request failed')

lib/drives/array-index.js

@@ -0,0 +1,28 @@
+class ArrayIndex {
+  constructor () {
+    this._arr = []
+  }
+
+  _getFreeIndex () {
+    var idx = this._arr.indexOf(null)
+    if (idx === -1) idx = this._arr.length
+    if (!idx) idx = 1
+    return idx
+  }
+
+  get (idx) {
+    return this._arr[idx]
+  }
+
+  insert (value) {
+    const idx = this._getFreeIndex()
+    this._arr[idx] = value
+    return idx
+  }
+
+  delete (idx) {
+    this._arr[idx] = null
+  }
+}
+
+module.exports = ArrayIndex

lib/drives/index.js

@@ -15,9 +15,12 @@ const {
   toStat,
   fromMount,
   toDriveStats,
+  toDownloadProgress,
+  toDiffEntry,
   toChunks
 } = require('hyperdrive-daemon-client/lib/common')
 const { rpc } = require('hyperdrive-daemon-client')
+const ArrayIndex = require('./array-index.js')
 
 const log = require('../log').child({ component: 'drive-manager' })
 
@@ -40,9 +43,9 @@ class DriveManager extends EventEmitter {
 
     // TODO: Replace with an LRU cache.
     this._drives = new Map()
-    this._sessions = new Map()
     this._watchers = new Map()
-    this._sessionCounter = 0
+    this._sessions = new ArrayIndex()
+    this._downloads = new ArrayIndex()
 
     this._readyPromise = null
 
@@ -123,12 +126,12 @@ class DriveManager extends EventEmitter {
 
   async createSession (key, opts) {
     const drive = await this.get(key, opts)
-    this._sessions.set(++this._sessionCounter, drive)
-    return { drive, session: this._sessionCounter }
+    const sessionId = this._sessions.insert(drive)
+    return { drive, session: sessionId }
   }
 
-  async closeSession (id) {
-    this._sessions.delete(id)
+  closeSession (id) {
+    return this._sessions.delete(id)
   }
 
   async getAllStats () {
@@ -198,7 +201,8 @@ class DriveManager extends EventEmitter {
     keyString = this._generateKeyString(key, opts)
 
     if (drive.writable) {
-      await this._configureDrive(drive, opts && opts.configure)
+      // TODO: Replace everything in _configureDrive with virtual files in the FUSE component.
+      // await this._configureDrive(drive, opts && opts.configure)
     } else {
       // All read-only drives are currently published by default.
       await this.publish(drive)
@@ -230,6 +234,18 @@ class DriveManager extends EventEmitter {
 
   getHandlers () {
     return {
+      version: async (call) => {
+        const id = call.request.getId()
+
+        if (!id) throw new Error('A version request must specify a session ID.')
+        const drive = this.driveForSession(id)
+
+        const rsp = new rpc.drive.messages.DriveVersionResponse()
+        rsp.setVersion(drive.version)
+
+        return rsp
+      },
+
       get: async (call) => {
         var driveOpts = call.request.getOpts()
         if (driveOpts) driveOpts = fromHyperdriveOptions(driveOpts)
@@ -293,6 +309,139 @@ class DriveManager extends EventEmitter {
         return rsp
       },
 
+      download: async (call) => {
+        const id = call.request.getId()
+        const path = call.request.getPath()
+        const detailed = call.request.getDetailed()
+
+        if (!id) throw new Error('A download request must specify a session ID.')
+        const drive = this.driveForSession(id)
+        var downloadId = null
+        var ended = false
+
+        const dl = drive.download(path, { detailed }, () => {
+          if (ended) return null
+          return finish(true)
+        })
+        downloadId = this._downloads.insert(dl)
+
+        call.on('end', cleanup)
+        call.on('close', cleanup)
+        call.on('finish', cleanup)
+        call.on('error', cleanup)
+
+        dl.on('start', start)
+        dl.on('cancelled', () => finish(false))
+        dl.on('progress', progress)
+
+        function progress (path, fileStats, totals) {
+          const rsp = new rpc.drive.messages.DownloadResponse()
+          rsp.setDownloadid(downloadId)
+          rsp.setType(rpc.drive.messages.DownloadResponse.Type.PROGRESS)
+
+          if (detailed) {
+            const fileRsps = fileStats.map(getFileStatusRsp)
+            rsp.setFilesList(fileRsps)
+          }
+
+          const progressRsp = toDownloadProgress(totals)
+          rsp.setProgress(progressRsp)
+
+          call.write(rsp)
+        }
+
+        function start (files, totals) {
+          const rsp = new rpc.drive.messages.DownloadResponse()
+          rsp.setDownloadid(downloadId)
+          rsp.setType(rpc.drive.messages.DownloadResponse.Type.START)
+
+          if (detailed) {
+            const fileRsps = files.map(getFileStatusRsp)
+            rsp.setFilesList(fileRsps)
+          }
+
+          const progressRsp = toDownload(totals)
+          rsp.setProgress(progressRsp)
+
+          call.write(rsp)
+        }
+
+        function finish (completed) {
+          ended = true
+          const rsp = new rpc.drive.messages.DownloadResponse()
+          rsp.setDownloadid(downloadId)
+          rsp.setType(rpc.drive.messages.DownloadResponse.Type.FINISH)
+
+          const progress = new rpc.drive.messages.DownloadProgress()
+          progress.setCompleted(!!completed)
+          progress.setCancelled(!completed)
+          rsp.setProgress(progress)
+
+          cleanup()
+          call.end(rsp)
+        }
+
+        function cleanup () {
+          this._downloads.delete(downloadId)
+          dl.removeAllListeners('progress')
+          dl.removeAllListeners('cancelled')
+          dl.removeAllListeners('start')
+        }
+
+        function getFileStatusRsp (path, stats) {
+          const fileStatus = new rpc.drive.messages.FileDownloadStatus()
+          fileStatus.setPath(path)
+          if (stats) fileStatus.setProgress(toDownloadProgress(stats))
+          return fileStatus
+        }
+      },
+
+      undownload: async (call) => {
+        const id = call.request.getId()
+        const downloadId = call.request.getDownloadid()
+
+        if (!id) throw new Error('An undownload request must specify a session ID.')
+        if (!downloadId) throw new Error('An undownload request must specify a download ID.')
+        const drive = this.driveForSession(id)
+
+        const dl = this._downloads.get(downloadId)
+        if (dl) dl.cancel()
+        this._downloads.delete(downloadId)
+
+        return new rpc.drive.messages.UndownloadResponse()
+      },
+
+      createDiffStream: async (call) => {
+        const id = call.request.getId()
+        const prefix = call.request.getPrefix()
+        const otherVersion = call.request.getOther()
+
+        if (!id) throw new Error('A diff stream request must specify a session ID.')
+        const drive = this.driveForSession(id)
+
+        const stream = drive.createDiffStream(otherVersion, prefix)
+
+        const rspMapper = map.obj(chunk => {
+          const rsp = new rpc.drive.messages.DiffStreamResponse()
+          if (!chunk) return rsp
+
+          const { name, type, value } = chunk
+          rsp.setType(type)
+          rsp.setName(name)
+          if (type === 'put') {
+            rsp.setValue(toDiffEntry({ stat: value }))
+          } else {
+            rsp.setValue(toDiffEntry({ mount: value }))
+          }
+
+          return rsp
+        })
+
+        pump(stream, rspMapper, call, err => {
+          if (err) log.error({ id, err }, 'createDiffStream error')
+        })
+      },
+
       createReadStream: async (call) => {
         const id = call.request.getId()
         const path = call.request.getPath()
@@ -408,8 +557,10 @@ class DriveManager extends EventEmitter {
         if (!path) throw new Error('A stat request must specify a path. ')
         const drive = this.driveForSession(id)
 
+        const method = lstat ? drive.lstat.bind(drive) : drive.stat.bind(drive)
+
         return new Promise((resolve, reject) => {
-          drive.stat(path, { followLink: lstat }, (err, stat) => {
+          method(path, (err, stat) => {
             if (err) return reject(err)
 
             const rsp = new rpc.drive.messages.StatResponse()

package.json

@@ -36,7 +36,7 @@
     "forever-monitor": "^1.7.1",
     "fs-extra": "^7.0.1",
     "google-protobuf": "^3.8.0",
-    "hyperdrive": "^10.0.0-rc10",
+    "hyperdrive": "^10.2.0",
     "hyperdrive-daemon-client": "^0.10.1",
     "level": "^4.0.0",
     "mkdirp": "^0.5.1",

test/hyperdrive.js

@@ -11,6 +11,8 @@ test('can write/read a file from a remote hyperdrive', async t => {
     t.true(drive.key)
     t.same(drive.id, 1)
 
+    const version = await drive.version()
+
     await drive.writeFile('hello', 'world')
 
     const contents = await drive.readFile('hello')
@@ -122,11 +124,10 @@ test('can list a directory from a remote hyperdrive', async t => {
     await drive.writeFile('adios', 'amigo')
 
     const files = await drive.readdir('')
-    t.same(files.length, 4)
+    t.same(files.length, 3)
     t.notEqual(files.indexOf('hello'), -1)
     t.notEqual(files.indexOf('goodbye'), -1)
     t.notEqual(files.indexOf('adios'), -1)
-    t.notEqual(files.indexOf('.key'), -1)
 
     await drive.close()
   } catch (err) {
@@ -137,6 +138,70 @@ test('can list a directory from a remote hyperdrive', async t => {
   t.end()
 })
 
+test('can create a diff stream on a remote hyperdrive', async t => {
+  const { client, cleanup } = await createOne()
+
+  try {
+    const drive1 = await client.drive.get()
+    const drive2 = await client.drive.get()
+
+    await drive1.writeFile('hello', 'world')
+    const v1 = await drive1.version()
+    await drive1.writeFile('goodbye', 'dog')
+    const v2 = await drive1.version()
+    await drive1.mount('d2', { key: drive2.key })
+    const v3 = await drive1.version()
+    await drive1.unmount('d2')
+
+    const diff1 = await drive1.createDiffStream()
+    const checkout = await drive1.checkout(v2)
+    const diff2 = await checkout.createDiffStream(v1)
+    const diff3 = await drive1.createDiffStream(v3)
+
+    await validate(diff1, [
+      { type: 'put', name: 'goodbye' },
+      { type: 'put', name: 'hello' }
+    ])
+    await validate(diff2, [
+      { type: 'put', name: 'goodbye'}
+    ])
+    await validate(diff3, [
+      // TODO: The first is a false positive.
+      { type: 'put', name: 'goodbye' },
+      { type: 'unmount', name: 'd2' }
+    ])
+
+    await drive1.close()
+    await drive2.close()
+  } catch (err) {
+    t.fail(err)
+  }
+
+  await cleanup()
+  t.end()
+
+  async function validate (stream, expected) {
+    return new Promise((resolve, reject) => {
+      var seen = 0
+      stream.on('end', () => {
+        t.same(seen, expected.length)
+        return resolve()
+      })
+      stream.on('error', t.fail.bind(t))
+      stream.on('data', ({ name, left, right }) => {
+        t.same(name, expected[seen].name)
+        if (left) {
+          t.same(left.type, expected[seen].left.type)
+        }
+        if (right) {
+          t.same(right.type, expected[seen].right.type)
+        }
+        seen++
+      })
+    })
+  }
+})
+
 test('can read/write multiple remote hyperdrives on one server', async t => {
   const { client, cleanup } = await createOne()
   var startingId = 1
@@ -311,6 +376,9 @@ test('can create a symlink to directories', async t => {
     t.same(files.length, 1)
     t.same(files[0], 'world')
 
+    const stat = await drive.lstat('other_world')
+    t.true(stat.isSymbolicLink())
+
     await drive.close()
   } catch (err) {
     t.fail(err)