Support abstract-level v3 APIs

guest.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { AbstractLevel, AbstractIterator } = require('abstract-level')
+const { AbstractLevel, AbstractIterator, AbstractSnapshot } = require('abstract-level')
 const lpstream = require('@vweevers/length-prefixed-stream')
 const ModuleError = require('module-error')
 const { input, output } = require('./tags')
@@ -18,35 +18,46 @@ const kRef = Symbol('ref')
 const kDb = Symbol('db')
 const kRequests = Symbol('requests')
 const kIterators = Symbol('iterators')
+const kSnapshots = Symbol('snapshots')
 const kRetry = Symbol('retry')
 const kRpcStream = Symbol('rpcStream')
 const kFlushed = Symbol('flushed')
 const kWrite = Symbol('write')
+const kGetSync = Symbol('getSync')
 const kRequest = Symbol('request')
 const kPending = Symbol('pending')
 const kCallback = Symbol('callback')
 const kSeq = Symbol('seq')
 const kErrored = Symbol('errored')
-const noop = function () {}
+const kAbortIterator = Symbol('abortIterator')
+const kOnDbClosing = Symbol('onDbClosing')
+const kId = Symbol('id')
 
 class ManyLevelGuest extends AbstractLevel {
   constructor (options) {
-    const { retry, _remote, ...forward } = options || {}
+    const { retry, _remote, getSync, ...forward } = options || {}
+    const hasGetSync = getSync === true || typeof getSync === 'function'
 
     super({
       encodings: { buffer: true },
       snapshots: !retry,
+      implicitSnapshots: !retry,
+      explicitSnapshots: !retry,
       permanence: true,
       seek: true,
+      has: true,
+      getSync: hasGetSync,
       createIfMissing: false,
       errorIfExists: false
     }, forward)
 
     this[kIterators] = new IdMap()
     this[kRequests] = new IdMap()
+    this[kSnapshots] = new IdMap()
     this[kRetry] = !!retry
     this[kEncode] = lpstream.encode()
     this[kRemote] = _remote || null
+    this[kGetSync] = typeof getSync === 'function' ? getSync : null
     this[kCleanup] = null
     this[kRpcStream] = null
     this[kRef] = null
@@ -105,6 +116,14 @@ class ManyLevelGuest extends AbstractLevel {
         case output.getManyCallback:
           ongetmanycallback(res)
           break
+
+        case output.hasCallback:
+          onhascallback(res)
+          break
+
+        case output.hasManyCallback:
+          onhasmanycallback(res)
+          break
       }
 
       self[kFlushed]()
@@ -167,6 +186,20 @@ class ManyLevelGuest extends AbstractLevel {
       if (res.error) req.callback(new ModuleError('Could not get values', { code: res.error }))
       else req.callback(null, res.values.map(v => normalizeValue(v.value)))
     }
+
+    function onhascallback (res) {
+      const req = self[kRequests].remove(res.id)
+      if (!req || !req.callback) return
+      if (res.error) req.callback(new ModuleError('Could not check key', { code: res.error }))
+      else req.callback(null, res.value)
+    }
+
+    function onhasmanycallback (res) {
+      const req = self[kRequests].remove(res.id)
+      if (!req || !req.callback) return
+      if (res.error) req.callback(new ModuleError('Could not check keys', { code: res.error }))
+      else req.callback(null, res.values)
+    }
   }
 
   // Alias for backwards compat with multileveldown
@@ -204,17 +237,7 @@ class ManyLevelGuest extends AbstractLevel {
     }
 
     for (const req of this[kIterators].clear()) {
-      // Cancel in-flight operation if any
-      // TODO: does this need to be refactored to use AbortError to pass back up to the request initiator?
-      const callback = req.iterator[kCallback]
-      req.iterator[kCallback] = null
-
-      if (callback) {
-        callback(new ModuleError(msg, { code }))
-      }
-
-      // Note: an in-flight operation would block close()
-      req.iterator.close(noop)
+      req.iterator[kAbortIterator](code)
     }
   }
 
@@ -227,6 +250,7 @@ class ManyLevelGuest extends AbstractLevel {
         tag: input.get,
         id: 0,
         key: key,
+        snapshot: snapshotId(opts.snapshot),
         // This will resolve or reject based on the Host's response
         callback: (err, value) => {
           if (err) reject(err)
@@ -239,6 +263,18 @@ class ManyLevelGuest extends AbstractLevel {
     })
   }
 
+  _getSync (key, opts) {
+    if (this[kDb]) return this[kDb]._getSync(key, opts)
+
+    if (this[kGetSync]) {
+      return normalizeValue(this[kGetSync](key, encodeOptionsSnapshot(opts)))
+    }
+
+    throw new ModuleError('Database does not support getSync()', {
+      code: 'LEVEL_NOT_SUPPORTED'
+    })
+  }
+
   async _getMany (keys, opts) {
     if (this[kDb]) return this[kDb]._getMany(keys, opts)
 
@@ -247,6 +283,7 @@ class ManyLevelGuest extends AbstractLevel {
         tag: input.getMany,
         id: 0,
         keys: keys,
+        snapshot: snapshotId(opts.snapshot),
         // This will resolve or reject based on the Host's response
         callback: (err, values) => {
           if (err) reject(err)
@@ -259,6 +296,46 @@ class ManyLevelGuest extends AbstractLevel {
     })
   }
 
+  async _has (key, opts) {
+    if (this[kDb]) return this[kDb]._has(key, opts)
+
+    return new Promise((resolve, reject) => {
+      const req = {
+        tag: input.has,
+        id: 0,
+        key: key,
+        snapshot: snapshotId(opts.snapshot),
+        callback: (err, value) => {
+          if (err) reject(err)
+          else resolve(value)
+        }
+      }
+
+      req.id = this[kRequests].add(req)
+      this[kWrite](req)
+    })
+  }
+
+  async _hasMany (keys, opts) {
+    if (this[kDb]) return this[kDb]._hasMany(keys, opts)
+
+    return new Promise((resolve, reject) => {
+      const req = {
+        tag: input.hasMany,
+        id: 0,
+        keys: keys,
+        snapshot: snapshotId(opts.snapshot),
+        callback: (err, values) => {
+          if (err) reject(err)
+          else resolve(values)
+        }
+      }
+
+      req.id = this[kRequests].add(req)
+      this[kWrite](req)
+    })
+  }
+
   async _put (key, value, opts) {
     if (this[kDb]) return this[kDb]._put(key, value, opts)
 
@@ -323,6 +400,8 @@ class ManyLevelGuest extends AbstractLevel {
   async _clear (opts) {
     if (this[kDb]) return this[kDb]._clear(opts)
 
+    opts = encodeOptionsSnapshot(opts)
+
     return new Promise((resolve, reject) => {
       const req = {
         tag: input.clear,
@@ -340,6 +419,22 @@ class ManyLevelGuest extends AbstractLevel {
     })
   }
 
+  _snapshot (options) {
+    if (this[kRetry]) {
+      throw new ModuleError('Database does not support explicit snapshots', {
+        code: 'LEVEL_NOT_SUPPORTED'
+      })
+    }
+
+    if (this[kDb]) return this[kDb].snapshot(options)
+
+    const snapshot = new ManyLevelGuestSnapshot(this, options)
+    const req = { tag: input.snapshot, id: snapshot[kId] }
+
+    this[kWrite](req)
+    return snapshot
+  }
+
   async [kWrite] (req) {
     if (this[kRequests].size + this[kIterators].size === 1) ref(this[kRef])
     const enc = input.encoding(req.tag)
@@ -377,13 +472,14 @@ class ManyLevelGuest extends AbstractLevel {
       // For tests only so does not need error handling
       this[kExplicitClose] = false
       const remote = this[kRemote]()
+      const local = this.createRpcStream()
       pipeline(
         remote,
-        this.createRpcStream(),
+        local,
         remote
       ).catch(err => {
         if (err.code === 'ABORT_ERR') {
-          return this.close()
+          if (!this[kExplicitClose] && this[kRpcStream] === local) return this.close()
         }
       })
     } else if (this[kExplicitClose]) {
@@ -422,16 +518,23 @@ class ManyLevelGuestIterator extends AbstractIterator {
     this[kPending] = []
     this[kCallback] = null
     this[kSeq] = 0
+    this[kOnDbClosing] = () => {
+      if (this.db[kRpcStream] === null && this.db[kDb] === null) {
+        this[kAbortIterator]('LEVEL_ITERATOR_NOT_OPEN')
+      }
+    }
+    db.once('closing', this[kOnDbClosing])
 
     const req = this[kRequest] = {
       tag: input.iterator,
       id: 0,
       seq: 0,
       iterator: this,
-      options,
+      options: encodeOptionsSnapshot(options),
       consumed: 0,
       bookmark: null,
-      seek: null
+      seek: null,
+      snapshot: snapshotId(options.snapshot)
     }
 
     const ack = this[kAckMessage] = {
@@ -529,10 +632,45 @@ class ManyLevelGuestIterator extends AbstractIterator {
   }
 
   async _close () {
+    this.db.removeListener('closing', this[kOnDbClosing])
     await this.db[kWrite]({ tag: input.iteratorClose, id: this[kRequest].id })
     this.db[kIterators].remove(this[kRequest].id)
     this.db[kFlushed]()
   }
+
+  [kAbortIterator] (code) {
+    this[kPending].push({ error: code })
+    this[kEnded] = true
+
+    const callback = this[kCallback]
+    this[kCallback] = null
+
+    if (callback) callback(null)
+  }
+}
+
+class ManyLevelGuestSnapshot extends AbstractSnapshot {
+  constructor (db, options) {
+    super(options)
+    this.db = db
+    this[kId] = db[kSnapshots].add(this)
+  }
+
+  async _close () {
+    this.db[kSnapshots].remove(this[kId])
+    await this.db[kWrite]({ tag: input.snapshotClose, id: this[kId] })
+    this.db[kFlushed]()
+  }
+}
+
+function snapshotId (snapshot) {
+  return snapshot instanceof ManyLevelGuestSnapshot ? snapshot[kId] : 0
+}
+
+function encodeOptionsSnapshot (options) {
+  if (!options || !options.snapshot) return options
+  const snapshot = snapshotId(options.snapshot)
+  return snapshot ? Object.assign({}, options, { snapshot }) : options
 }
 
 function normalizeValue (value) {