Added readFile/writeFile chunks + unmount + tests for both

Author: andrewosh

lib/drives/index.js

@@ -14,7 +14,8 @@ const {
   toStat,
   toMount,
   fromMount,
-  toDriveStats
+  toDriveStats,
+  toChunks
 } = require('hyperdrive-daemon-client/lib/common')
 const { rpc } = require('hyperdrive-daemon-client')
 
@@ -316,43 +317,63 @@ function createDriveHandlers (driveManager) {
       return rsp
     },
 
+    createReadStream: async (call) => {
+
+    },
+
     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.')
+      if (!path) throw new Error('A readFile request must specify a path.')
       const drive = driveManager.driveForSession(id)
 
-      return new Promise((resolve, reject) => {
+      const content = await 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)
+          return resolve(content)
         })
       })
+
+      const chunks = toChunks(content)
+      for (const chunk of chunks) {
+        const rsp = new rpc.drive.messages.ReadFileResponse()
+        rsp.setChunk(chunk)
+        call.write(rsp)
+      }
+      call.end()
     },
 
-    writeFile: async (call) => {
-      const id = call.request.getId()
-      const path = call.request.getPath()
-      const contents = Buffer.from(call.request.getContent())
+    createWriteStream: async (call) => {
 
-      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)
+    },
 
+    writeFile: async (call) => {
       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)
+        call.once('data', req => {
+          const id = req.getId()
+          const path = req.getPath()
+
+          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.')
+          const drive = driveManager.driveForSession(id)
+
+          return loadContent(resolve, reject, path, drive)
         })
       })
+
+      function loadContent (resolve, reject, path, drive) {
+        return collectStream(call, (err, reqs) => {
+          if (err) return reject(err)
+          const chunks = reqs.map(req => Buffer.from(req.getChunk()))
+          return drive.writeFile(path, Buffer.concat(chunks), err => {
+            if (err) return reject(err)
+            const rsp = new rpc.drive.messages.WriteFileResponse()
+            return resolve(rsp)
+          })
+        })
+      }
     },
 
     stat: async (call) => {
@@ -454,8 +475,10 @@ function createDriveHandlers (driveManager) {
 
     mount: async (call) => {
       const id = call.request.getId()
-      const path = call.request.getPath()
-      const opts = fromMount(call.request.getOpts())
+      const mountInfo = call.request.getInfo()
+
+      const path = mountInfo.getPath()
+      const opts = fromMount(mountInfo.getOpts())
 
       if (!id) throw new Error('A mount request must specify a session ID.')
       if (!path) throw new Error('A mount request must specify a path.')
@@ -471,6 +494,23 @@ function createDriveHandlers (driveManager) {
       })
     },
 
+    unmount: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+
+      if (!id) throw new Error('An unmount request must specify a session ID.')
+      if (!path) throw new Error('An unmount request must specify a path.')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.unmount(path, err => {
+          if (err) return reject(err)
+          const rsp = new rpc.drive.messages.UnmountDriveResponse()
+          return resolve(rsp)
+        })
+      })
+    },
+
     watch: async (call) => {
       const id = call.request.getId()
       var path = call.request.getPath()
@@ -487,7 +527,11 @@ function createDriveHandlers (driveManager) {
       call.on('finish', onclose)
       call.on('error', onclose)
 
+      var closed = false
       function onclose (err) {
+        if (closed) return
+        closed = true
+
         watcher.destroy()
         log.debug({ id, path }, 'unregistering watcher')
         if (err && !call.destroyed) call.destroy(err)

test/hyperdrive.js

@@ -24,6 +24,30 @@ test('can write/read a file from a remote hyperdrive', async t => {
   t.end()
 })
 
+test('can write/read a large file from a remote hyperdrive', async t => {
+  const { client, cleanup } = await createOne()
+
+  const content = Buffer.alloc(3.9e6 * 10.11).fill('abcdefghi')
+
+  try {
+    const { opts, id } = await client.drive.get()
+    t.true(opts.key)
+    t.same(id, 1)
+
+    await client.drive.writeFile(id, 'hello', content)
+
+    const contents = await client.drive.readFile(id, 'hello')
+    t.same(contents, content)
+
+    await client.drive.close(id)
+  } catch (err) {
+    t.fail(err)
+  }
+
+  await cleanup()
+  t.end()
+})
+
 test('can stat a file from a remote hyperdrive', async t => {
   const { client, cleanup } = await createOne()
 
@@ -143,3 +167,49 @@ test('can mount a drive within a remote hyperdrive', async t => {
   await cleanup()
   t.end()
 })
+
+test.only('can unmount a drive within a remote hyperdrive', async t => {
+  const { client, cleanup } = await createOne()
+
+  try {
+    const { opts: opts1, id: id1 } = await client.drive.get()
+    t.true(opts1.key)
+    t.same(id1, 1)
+
+    const { opts: opts2, id: id2 } = await client.drive.get()
+    t.true(opts2.key)
+    t.same(id2, 2)
+    t.notEqual(opts1.key, opts2.key)
+
+    const noVersion = { ...opts2, version: null }
+
+    await client.drive.mount(id1, 'a', noVersion)
+
+    await client.drive.writeFile(id1, 'a/hello', 'world')
+    await client.drive.writeFile(id1, 'a/goodbye', 'dog')
+    await client.drive.writeFile(id1, 'adios', 'amigo')
+    await client.drive.writeFile(id2, 'hamster', 'wheel')
+
+    t.same(await client.drive.readFile(id1, 'adios'), Buffer.from('amigo'))
+    t.same(await client.drive.readFile(id1, 'a/hello'), Buffer.from('world'))
+    t.same(await client.drive.readFile(id2, 'hello'), Buffer.from('world'))
+    t.same(await client.drive.readFile(id2, 'hamster'), Buffer.from('wheel'))
+
+    await client.drive.unmount(id1, 'a')
+    try {
+      await client.drive.readFile(id1, 'a/hello')
+    } catch (err) {
+      t.true(err)
+      t.same(err.code, 2)
+    }
+
+    await client.drive.close(id1)
+    await client.drive.close(id2)
+  } catch (err) {
+    t.fail(err)
+  }
+
+  await cleanup()
+  t.end()
+
+})