chore: merge zcstarr:master

Author: achingbrain

package.json

@@ -39,16 +39,13 @@
   },
   "homepage": "https://github.com/ipfs/js-datastore-core#readme",
   "devDependencies": {
-    "aegir": "^18.2.2",
+    "aegir": "^v18.2.0",
     "chai": "^4.2.0",
     "dirty-chai": "^2.0.1",
-    "flow-bin": "~0.98.1"
+    "flow-bin": "~0.83.0"
   },
   "dependencies": {
-    "async": "^2.6.1",
-    "interface-datastore": "~0.6.0",
-    "pull-many": "^1.0.8",
-    "pull-stream": "^3.6.9"
+    "interface-datastore": "~0.7.0"
   },
   "engines": {
     "node": ">=6.0.0",

src/keytransform.js

@@ -1,7 +1,8 @@
 /* @flow */
 'use strict'
 
-const pull = require('pull-stream')
+const utils = require('interface-datastore').utils
+const map = utils.map
 
 /* ::
 import type {Key, Datastore, Batch, Query, QueryResult, Callback} from 'interface-datastore'
@@ -38,24 +39,24 @@ class KeyTransformDatastore /* :: <Value> */ {
     this.transform = transform
   }
 
-  open (callback /* : Callback<void> */) /* : void */ {
-    this.child.open(callback)
+  open () /* : Promise<void> */ {
+    return this.child.open()
   }
 
-  put (key /* : Key */, val /* : Value */, callback /* : Callback<void> */) /* : void */ {
-    this.child.put(this.transform.convert(key), val, callback)
+  async put (key /* : Key */, val /* : Value */) /* : Promise<void> */ {
+    return this.child.put(this.transform.convert(key), val)
   }
 
-  get (key /* : Key */, callback /* : Callback<Value> */) /* : void */ {
-    this.child.get(this.transform.convert(key), callback)
+  async get (key /* : Key */) /* : Promise<Value> */ {
+    return this.child.get(this.transform.convert(key))
   }
 
-  has (key /* : Key */, callback /* : Callback<bool> */) /* : void */ {
-    this.child.has(this.transform.convert(key), callback)
+  async has (key /* : Key */) /* : Promise<bool> */ {
+    return this.child.has(this.transform.convert(key))
   }
 
-  delete (key /* : Key */, callback /* : Callback<void> */) /* : void */ {
-    this.child.delete(this.transform.convert(key), callback)
+  async delete (key /* : Key */) /* : Promise<void> */ {
+    return this.child.delete(this.transform.convert(key))
   }
 
   batch () /* : Batch<Value> */ {
@@ -67,24 +68,21 @@ class KeyTransformDatastore /* :: <Value> */ {
       delete: (key /* : Key */) /* : void */ => {
         b.delete(this.transform.convert(key))
       },
-      commit: (callback /* : Callback<void> */) /* : void */ => {
-        b.commit(callback)
+      commit: async () /* : Promise<void> */ => {
+        return b.commit()
       }
     }
   }
 
-  query (q /* : Query<Value> */) /* : QueryResult<Value> */ {
-    return pull(
-      this.child.query(q),
-      pull.map(e => {
-        e.key = this.transform.invert(e.key)
-        return e
-      })
-    )
+  query (q /* : Query<Value> */) /* : Iterator */ {
+    return map(this.child.query(q), e => {
+      e.key = this.transform.invert(e.key)
+      return e
+    })
   }
 
-  close (callback /* : Callback<void> */) /* : void */ {
-    this.child.close(callback)
+  async close () /* : Promise<void> */ {
+    return this.child.close()
   }
 }
 

src/mount.js

@@ -1,15 +1,12 @@
 /* @flow */
 'use strict'
 
-const each = require('async/each')
-const many = require('pull-many')
-const pull = require('pull-stream')
-
 const Key = require('interface-datastore').Key
 const Errors = require('interface-datastore').Errors
 const utils = require('interface-datastore').utils
-const asyncFilter = utils.asyncFilter
-const asyncSort = utils.asyncSort
+const filter = utils.filter
+const take = utils.take
+const sortAll = utils.sortAll
 const replaceStartWith = utils.replaceStartWith
 
 const Keytransform = require('./keytransform')
@@ -34,10 +31,8 @@ class MountDatastore /* :: <Value> */ {
     this.mounts = mounts.slice()
   }
 
-  open (callback /* : Callback<void> */) /* : void */ {
-    each(this.mounts, (m, cb) => {
-      m.datastore.open(cb)
-    }, callback)
+  async open () /* : Promise<void> */ {
+    return Promise.all(this.mounts.map((m) => m.datastore.open()))
   }
 
   /**
@@ -60,53 +55,44 @@ class MountDatastore /* :: <Value> */ {
     }
   }
 
-  put (key /* : Key */, value /* : Value */, callback /* : Callback<void> */) /* : void */ {
+  async put (key /* : Key */, value /* : Value */) /* : Promise<void> */ {
     const match = this._lookup(key)
     if (match == null) {
-      return callback(
-        Errors.dbWriteFailedError(new Error('No datastore mounted for this key'))
-      )
+      throw Errors.dbWriteFailedError(new Error('No datastore mounted for this key'))
     }
 
-    match.datastore.put(match.rest, value, callback)
+    return match.datastore.put(match.rest, value)
   }
 
-  get (key /* : Key */, callback /* : Callback<Value> */) /* : void */ {
+  async get (key /* : Key */) /* : Promise<Value> */ {
     const match = this._lookup(key)
     if (match == null) {
-      return callback(
-        Errors.notFoundError(new Error('No datastore mounted for this key'))
-      )
+      throw Errors.notFoundError(new Error('No datastore mounted for this key'))
     }
-
-    match.datastore.get(match.rest, callback)
+    return match.datastore.get(match.rest)
   }
 
-  has (key /* : Key */, callback /* : Callback<bool> */) /* : void */ {
+  async has (key /* : Key */) /* : Promise<bool> */ {
     const match = this._lookup(key)
     if (match == null) {
-      callback(null, false)
-      return
+      return false
     }
-
-    match.datastore.has(match.rest, callback)
+    return match.datastore.has(match.rest)
   }
 
-  delete (key /* : Key */, callback /* : Callback<void> */) /* : void */ {
+  async delete (key /* : Key */) /* : Promise<void> */ {
     const match = this._lookup(key)
     if (match == null) {
-      return callback(
-        Errors.dbDeleteFailedError(new Error('No datastore mounted for this key'))
-      )
+      throw Errors.dbDeleteFailedError(new Error('No datastore mounted for this key'))
     }
 
-    match.datastore.delete(match.rest, callback)
+    return match.datastore.delete(match.rest)
   }
 
-  close (callback /* : Callback<void> */) /* : void */ {
-    each(this.mounts, (m, cb) => {
-      m.datastore.close(cb)
-    }, callback)
+  async close () /* : Promise<void> */ {
+    return Promise.all(this.mounts.map((m) => {
+      return m.datastore.close()
+    }))
   }
 
   batch () /* : Batch<Value> */ {
@@ -137,10 +123,8 @@ class MountDatastore /* :: <Value> */ {
         const match = lookup(key)
         match.batch.delete(match.rest)
       },
-      commit: (callback /* : Callback<void> */) /* : void */ => {
-        each(Object.keys(batchMounts), (p, cb) => {
-          batchMounts[p].commit(cb)
-        }, callback)
+      commit: async () /* : Promise<void> */ => {
+        return Promise.all(Object.keys(batchMounts).map(p => batchMounts[p].commit()))
       }
     }
   }
@@ -168,27 +152,33 @@ class MountDatastore /* :: <Value> */ {
       })
     })
 
-    let tasks = [many(qs)]
-
-    if (q.filters != null) {
-      tasks = tasks.concat(q.filters.map(f => asyncFilter(f)))
-    }
-
-    if (q.orders != null) {
-      tasks = tasks.concat(q.orders.map(o => asyncSort(o)))
-    }
-
+    let it = _many(qs)
+    if (q.filters) q.filters.forEach(f => { it = filter(it, f) })
+    if (q.orders) q.orders.forEach(o => { it = sortAll(it, o) })
     if (q.offset != null) {
       let i = 0
-      tasks.push(pull.filter(() => i++ >= q.offset))
+      it = filter(it, () => i++ >= q.offset)
     }
+    if (q.limit != null) it = take(it, q.limit)
 
-    if (q.limit != null) {
-      tasks.push(pull.take(q.limit))
-    }
-
-    return pull.apply(null, tasks)
+    return it
   }
 }
 
+function _many (iterable) {
+  return (async function * () {
+    let completed = iterable.map(() => false)
+    while (!completed.every(Boolean)) {
+      for (const [idx, itr] of iterable.entries()) {
+        const it = await itr.next()
+        if (it.done) {
+          completed[idx] = true
+          continue
+        }
+        yield it.value
+      }
+    }
+  })()
+}
+
 module.exports = MountDatastore

src/namespace.js

@@ -41,7 +41,7 @@ class NamespaceDatastore/* :: <Value> */ extends KeytransformDatastore /* :: <Va
     this.prefix = prefix
   }
 
-  query (q /* : Query<Value> */)/* : QueryResult<Value> */ {
+  query (q /* : Query<Value> */)/* : Iterator */ {
     if (q.prefix && this.prefix.toString() !== '/') {
       return super.query(Object.assign({}, q, {
         prefix: this.prefix.child(new Key(q.prefix)).toString()

src/shard.js

@@ -121,24 +121,11 @@ function parseShardFun (str /* : string */) /* : ShardV1 */ {
   }
 }
 
-exports.readShardFun = (path /* : string */, store /* : Datastore<Buffer> */, callback /* : Callback<ShardV1> */) /* : void */ => {
+exports.readShardFun = async (path /* : string */, store /* : Datastore<Buffer> */) /* : Promise<ShardV1> */ => {
   const key = new Key(path).child(new Key(SHARDING_FN))
   const get = typeof store.getRaw === 'function' ? store.getRaw.bind(store) : store.get.bind(store)
-
-  get(key, (err, res) => {
-    if (err) {
-      return callback(err)
-    }
-
-    let shard
-    try {
-      shard = parseShardFun((res || '').toString().trim())
-    } catch (err) {
-      return callback(err)
-    }
-
-    callback(null, shard)
-  })
+  const res = await get(key)
+  return parseShardFun((res || '').toString().trim())
 }
 
 exports.readme = readme