feat: add streaming/cancellable API

package.json

@@ -35,14 +35,14 @@
   },
   "homepage": "https://github.com/ipfs/js-datastore-fs#readme",
   "dependencies": {
-    "datastore-core": "^1.0.0",
-    "fast-write-atomic": "~0.2.0",
+    "datastore-core": "^1.1.0",
+    "fast-write-atomic": "^0.2.0",
     "glob": "^7.1.3",
-    "interface-datastore": "~0.8.3",
+    "interface-datastore": "^1.0.2",
     "mkdirp": "^1.0.4"
   },
   "devDependencies": {
-    "aegir": "^21.10.2",
+    "aegir": "^22.0.0",
     "async-iterator-all": "^1.0.0",
     "chai": "^4.2.0",
     "cids": "~0.8.0",

src/index.js

@@ -6,21 +6,17 @@ const mkdirp = require('mkdirp')
 const promisify = require('util').promisify
 const writeAtomic = promisify(require('fast-write-atomic'))
 const path = require('path')
-
-const filter = require('interface-datastore').utils.filter
-const take = require('interface-datastore').utils.take
-const map = require('interface-datastore').utils.map
-const sortAll = require('interface-datastore').utils.sortAll
-const IDatastore = require('interface-datastore')
+const {
+  Adapter, Key, Errors, utils: {
+    map
+  }
+} = require('interface-datastore')
 
 const noop = () => {}
 const fsAccess = promisify(fs.access || noop)
 const fsReadFile = promisify(fs.readFile || noop)
 const fsUnlink = promisify(fs.unlink || noop)
 
-const Key = IDatastore.Key
-const Errors = IDatastore.Errors
-
 async function writeFile (path, contents) {
   try {
     await writeAtomic(path, contents)
@@ -47,63 +43,30 @@ async function writeFile (path, contents) {
  * Keys need to be sanitized before use, as they are written
  * to the file system as is.
  */
-class FsDatastore {
+class FsDatastore extends Adapter {
   constructor (location, opts) {
+    super()
+
     this.path = path.resolve(location)
     this.opts = Object.assign({}, {
       createIfMissing: true,
       errorIfExists: false,
       extension: '.data'
     }, opts)
-
-    if (this.opts.createIfMissing) {
-      this._openOrCreate()
-    } else {
-      this._open()
-    }
   }
 
   open () {
-    this._openOrCreate()
-  }
-
-  /**
-   * Check if the path actually exists.
-   * @private
-   * @returns {void}
-   */
-  _open () {
-    if (!fs.existsSync(this.path)) {
-      throw Errors.notFoundError(new Error(`Datastore directory: ${this.path} does not exist`))
-    }
-
-    if (this.opts.errorIfExists) {
-      throw Errors.dbOpenFailedError(new Error(`Datastore directory: ${this.path} already exists`))
-    }
-  }
-
-  /**
-   * Create the directory to hold our data.
-   *
-   * @private
-   * @returns {void}
-   */
-  _create () {
-    mkdirp.sync(this.path, { fs: fs })
-  }
-
-  /**
-   * Tries to open, and creates if the open fails.
-   *
-   * @private
-   * @returns {void}
-   */
-  _openOrCreate () {
     try {
-      this._open()
+      if (!fs.existsSync(this.path)) {
+        throw Errors.notFoundError(new Error(`Datastore directory: ${this.path} does not exist`))
+      }
+
+      if (this.opts.errorIfExists) {
+        throw Errors.dbOpenFailedError(new Error(`Datastore directory: ${this.path} already exists`))
+      }
     } catch (err) {
-      if (err.code === 'ERR_NOT_FOUND') {
-        this._create()
+      if (err.code === 'ERR_NOT_FOUND' && this.opts.createIfMissing) {
+        mkdirp.sync(this.path, { fs: fs })
         return
       }
 
@@ -165,7 +128,7 @@ class FsDatastore {
   }
 
   /**
-   * Store the given value under the key.
+   * Store the given value under the key
    *
    * @param {Key} key
    * @param {Buffer} val
@@ -252,84 +215,27 @@ class FsDatastore {
     }
   }
 
-  /**
-   * Create a new batch object.
-   *
-   * @returns {Batch}
-   */
-  batch () {
-    const puts = []
-    const deletes = []
-    return {
-      put (key, value) {
-        puts.push({ key: key, value: value })
-      },
-      delete (key) {
-        deletes.push(key)
-      },
-      commit: () /* :  Promise<void> */ => {
-        return Promise.all(
-          puts
-            .map((put) => this.put(put.key, put.value))
-            .concat(
-              deletes.map((del) => this.delete(del))
-            )
-        )
-      }
-    }
-  }
-
-  /**
-   * Query the store.
-   *
-   * @param {Object} q
-   * @returns {Iterable}
-   */
-  query (q) {
+  async * _all (q) { // eslint-disable-line require-await
     // glob expects a POSIX path
     const prefix = q.prefix || '**'
     const pattern = path
       .join(this.path, prefix, '*' + this.opts.extension)
       .split(path.sep)
       .join('/')
     const files = glob.sync(pattern)
-    let it
+
     if (!q.keysOnly) {
-      it = map(files, async (f) => {
+      yield * map(files, async (f) => {
         const buf = await fsReadFile(f)
         return {
           key: this._decode(f),
           value: buf
         }
       })
     } else {
-      it = map(files, f => ({ key: this._decode(f) }))
-    }
-
-    if (Array.isArray(q.filters)) {
-      it = q.filters.reduce((it, f) => filter(it, f), it)
-    }
-
-    if (Array.isArray(q.orders)) {
-      it = q.orders.reduce((it, f) => sortAll(it, f), it)
-    }
-
-    if (q.offset != null) {
-      let i = 0
-      it = filter(it, () => i++ >= q.offset)
+      yield * map(files, f => ({ key: this._decode(f) }))
     }
-
-    if (q.limit != null) {
-      it = take(it, q.limit)
-    }
-
-    return it
   }
-
-  /**
-   * Close the store.
-   */
-  close () { }
 }
 
 module.exports = FsDatastore

test/index.spec.js

@@ -43,19 +43,23 @@ describe('FsDatastore', () => {
         () => new FsStore(dir)
       ).to.not.throw()
     })
+  })
 
+  describe('open', () => {
     it('createIfMissing: false - folder missing', () => {
       const dir = utils.tmpdir()
+      const store = new FsStore(dir, { createIfMissing: false })
       expect(
-        () => new FsStore(dir, { createIfMissing: false })
+        () => store.open()
       ).to.throw()
     })
 
     it('errorIfExists: true - folder exists', () => {
       const dir = utils.tmpdir()
       mkdirp.sync(dir)
+      const store = new FsStore(dir, { errorIfExists: true })
       expect(
-        () => new FsStore(dir, { errorIfExists: true })
+        () => store.open()
       ).to.throw()
     })
   })