Added drive API tests

Author: andrewosh

index.js

@@ -9,7 +9,7 @@ const grpc = require('grpc')
 
 const { rpc, loadMetadata } = require('hyperdrive-daemon-client')
 const corestore = require('random-access-corestore')
-const SwarmNetworker = require('megastore-swarm-networking')
+const SwarmNetworker = require('corestore-swarm-networking')
 
 const { DriveManager, createDriveHandlers } = require('./lib/drives')
 const { catchErrors, serverError, requestError } = require('./lib/errors')
@@ -63,7 +63,6 @@ class HyperdriveDaemon extends EventEmitter {
   _ready () {
     return Promise.all([
       this.db.open(),
-      this.megastore.ready(),
       this.networking.listen(),
       this.drives.ready(),
       this.fuse ? this.fuse.ready() : Promise.resolve()
@@ -72,15 +71,10 @@ class HyperdriveDaemon extends EventEmitter {
     })
   }
 
-  close () {
+  async close () {
     if (this._isClosed) return Promise.resolve()
-    return new Promise((resolve, reject) => {
-      this.megastore.close(err => {
-      if (err) return reject(err)
-        this._isClosed = true
-        return resolve()
-      })
-    })
+    if (this.networking) await this.networking.close()
+    this._isClosed = true
   }
 
   async cleanup () {
@@ -90,20 +84,23 @@ class HyperdriveDaemon extends EventEmitter {
   }
 }
 
-async function start () {
+module.exports = async function start (opts = {}) {
   const metadata = await new Promise((resolve, reject) => {
     loadMetadata((err, metadata) => {
       if (err) return reject(err)
       return resolve(metadata)
     })
   })
-  const storageRoot = argv.storage
+  const storageRoot = opts.storage || argv.storage
   await ensureStorage()
 
   const daemonOpts = {}
-  if (argv.bootstrap.length) {
-    daemonOpts.network = {
-      bootstrap: argv.bootstrap
+  const bootstrapOpts = opts.bootstrap || argv.bootstrap
+  if (bootstrapOpts.length) {
+    if (bootstrapOpts === false && bootstrapOpts[0] === 'false') {
+      daemonOpts.network = { bootstrap: false }
+    } else {
+      daemonOpts.network = { bootstrap: bootstrapOpts }
     }
   }
   const daemon = new HyperdriveDaemon(storageRoot, daemonOpts)
@@ -122,18 +119,22 @@ async function start () {
     ...wrap(metadata, createMainHandlers(server, daemon), { authenticate: true })
   })
 
-  server.bind(`0.0.0.0:${argv.port}`, grpc.ServerCredentials.createInsecure())
+
+  const port = opts.port || argv.port
+  server.bind(`0.0.0.0:${port}`, grpc.ServerCredentials.createInsecure())
   server.start()
-  log.info({ port: argv.port }, 'server listening')
+  log.info({ port: port }, 'server listening')
 
   process.once('SIGINT', cleanup)
   process.once('SIGTERM', cleanup)
   process.once('unhandledRejection', cleanup)
   process.once('uncaughtException', cleanup)
 
+  return cleanup
+
   async function cleanup () {
     await daemon.close()
-    server.tryDestroy()
+    server.forceShutdown()
   }
 
   function ensureStorage () {

lib/drives/index.js

@@ -7,7 +7,14 @@ const collectStream = require('stream-collector')
 const bjson = require('buffer-json-encoding')
 const datEncoding = require('dat-encoding')
 
-const { fromHyperdriveOptions, toHyperdriveOptions } = require('hyperdrive-daemon-client/lib/common')
+const {
+  fromHyperdriveOptions,
+  toHyperdriveOptions,
+  fromStat,
+  toStat
+} = require('hyperdrive-daemon-client/lib/common')
+const { rpc } = require('hyperdrive-daemon-client')
+
 const log = require('../log').child({ component: 'drive-manager' })
 
 class DriveManager extends EventEmitter {
@@ -30,6 +37,8 @@ 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._ready = new Promise(async resolve => {
       await this._reseed()
@@ -97,6 +106,22 @@ class DriveManager extends EventEmitter {
     return keyString
   }
 
+  driveForSession (sessionId) {
+    const drive = this._sessions.get(sessionId)
+    if (!drive) throw new Error('Session does not exist.')
+    return drive
+  }
+
+  async createSession (key, opts) {
+    const drive = await this.get(key, opts)
+    this._sessions.set(++this._sessionCounter, drive)
+    return { drive, session: this._sessionCounter }
+  }
+
+  async closeSession (id) {
+    this._sessions.delete(id)
+  }
+
   async get (key, opts) {
     key = (key instanceof Buffer) ? datEncoding.decode(key) : key
     var keyString = this._generateKeyString(key, opts)
@@ -121,7 +146,7 @@ class DriveManager extends EventEmitter {
     })
 
     key = datEncoding.encode(drive.key)
-    keyString = this._generateKeystring(key, opts)
+    keyString = this._generateKeyString(key, opts)
 
 
     if (drive.writable) {
@@ -192,12 +217,117 @@ function createDriveHandlers (driveManager) {
       var driveOpts = call.request.getOpts()
       if (driveOpts) driveOpts = fromHyperdriveOptions(driveOpts)
 
-      const drive = await driveManager.get(driveOpts.key, driveOpts)
+      const { drive, session } = await driveManager.createSession(driveOpts.key, driveOpts)
       driveOpts.key = drive.key
       driveOpts.version = drive.version
 
-      const rsp = new rpc.drive.messages.GetResponse()
-      rsp.setOpts(driveOpts)
+      const rsp = new rpc.drive.messages.GetDriveResponse()
+      rsp.setId(session)
+      rsp.setOpts(toHyperdriveOptions(driveOpts))
+
+      return rsp
+    },
+
+    readFile: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+
+      if (!id) throw new Error('A readFile request must specify a session ID.')
+      if (!path) throw new Error('A writeFile request must specify a path.')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.readFile(path, (err, content) => {
+          if (err) return reject(err)
+
+          const rsp = new rpc.drive.messages.ReadFileResponse()
+          rsp.setContent(content)
+
+          return resolve(rsp)
+        })
+      })
+    },
+
+    writeFile: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+      const contents = Buffer.from(call.request.getContent())
+
+      if (!id) throw new Error('A writeFile request must specify a session ID.')
+      if (!path) throw new Error('A writeFile request must specify a path.')
+      if (!contents) throw new Error('A writeFile request must specify contents.')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.writeFile(path, contents, (err) => {
+          if (err) return reject(err)
+          const rsp = new rpc.drive.messages.WriteFileResponse()
+          return resolve(rsp)
+        })
+      })
+    },
+
+    stat: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+      const lstat = call.request.getLstat()
+
+      if (!id) throw new Error('A stat request must specify a session ID.')
+      if (!path) throw new Error('A stat request must specify a path. ')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.stat(path, { followLink: lstat }, (err, stat) => {
+          if (err) return reject(err)
+
+          const rsp = new rpc.drive.messages.StatResponse()
+          rsp.setStat(toStat(stat))
+
+          return resolve(rsp)
+        })
+      })
+    },
+
+    readdir: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+      const recursive = call.request.getRecursive()
+
+      if (!id) throw new Error('A readdir request must specify a session ID.')
+      if (!path) throw new Error('A readdir request must specify a path. ')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.readdir(path, { recursive }, (err, files) => {
+          if (err) return reject(err)
+
+          const rsp = new rpc.drive.messages.ReadDirectoryResponse()
+          console.log('FILES:', files)
+          rsp.setFilesList(files)
+
+          return resolve(rsp)
+        })
+      })
+    },
+
+    watch: async (call) => {
+
+    },
+
+    listen: async (call) => {
+
+    },
+
+    unwatch: async (call) => {
+
+    },
+
+    closeSession: async (call) => {
+      const id = call.request.getId()
+
+      const drive = driveManager.driveForSession(id)
+      driveManager.closeSession(id)
+      const rsp = new rpc.drive.messages.CloseSessionResponse()
 
       return rsp
     }

lib/fuse/index.js

@@ -248,19 +248,16 @@ class FuseManager extends EventEmitter {
 
   async _driveForPath (path, opts = {}) {
     const self = this
-    console.error('GETTING DRIVE FOR PATH:', path)
     if (!this._rootDrive && path !== '/hyperdrive') throw new Error('You can only mount the root hyperdrive at /hyperdrive')
 
     if (!this._rootDrive) {
-      console.log('NO ROOT DRIVE')
       const drive = await this.driveManager.get(opts.key, { ...opts, configure: { rootDrive: true } })
       return { drive, root: true }
     }
 
     if (path.startsWith(this._rootMnt) && path !== this._rootMnt) {
       const relativePath = path.slice(this._rootMnt.length)
       if (!relativePath.startsWith('/home')) throw new Error('You can only mount sub-hyperdrives within the home directory.')
-      console.error('IT IS A SUBDRIVE, RELATIVE:', relativePath)
       return getSubdrive(relativePath)
     }
 
@@ -277,9 +274,7 @@ class FuseManager extends EventEmitter {
       })
       const drive = await self.driveManager.get(key, { ...opts })
       if (opts.key) {
-        console.error('PUBLISHING DRIVE BY KEY')
         await self.driveManager.publish(drive)
-        console.error('PUBLISHED')
       }
       return { drive, relativePath, root: false }
     }

lib/log.js

@@ -4,6 +4,6 @@ const argv = require('yargs').argv
 // Forever will redirect stdout to the correct log file.
 module.exports = pino({
   name: 'hyperdrive',
-  level: argv['log-level'],
+  level: argv['log-level'] || 'info',
   enabled: true,
 }, pino.destination(2))

package.json

@@ -25,6 +25,7 @@
   },
   "homepage": "https://github.com/andrewosh/hyperdrive-daemon#readme",
   "dependencies": {
+    "buffer-json-encoding": "^1.0.2",
     "chalk": "^2.4.2",
     "collect-stream": "^1.2.1",
     "dat-encoding": "^5.0.1",
@@ -37,9 +38,9 @@
     "hyperdrive-daemon-client": "^0.9.4",
     "hyperdrive-fuse": "^1.1.0",
     "level": "^4.0.0",
-    "mkdirp": "^0.5.1",
-    "mini-megastore": "^1.0.0",
     "megastore-swarm-networking": "^1.0.0",
+    "mini-megastore": "^1.0.0",
+    "mkdirp": "^0.5.1",
     "ora": "^3.4.0",
     "pino": "^5.12.6",
     "prettier-bytes": "^1.0.4",
@@ -51,8 +52,10 @@
   },
   "devDependencies": {
     "memdb": "^1.3.1",
+    "pify": "^4.0.1",
     "random-access-memory": "^3.1.1",
     "standard": "^12.0.1",
-    "tape": "^4.10.1"
+    "tape": "^4.10.1",
+    "tmp-promise": "^2.0.1"
   }
 }