WIP instead of having one function that can handle both async and callback signatures, let's implement a new Async suite of functions, and use callbackify to shim the old interface

pull/11/head
Timothée Rebours 3 years ago
parent 5a4859c3ba
commit f9e4c525e4
  1. 123
      lib/datastore.js

@ -1,4 +1,5 @@
const { EventEmitter } = require('events')
const { callbackify } = require('util')
const async = require('async')
const Cursor = require('./cursor.js')
const customUtils = require('./customUtils.js')
@ -267,17 +268,52 @@ class Datastore extends EventEmitter {
* @param {Boolean} dontExpireStaleDocs Optional, defaults to false, if true don't remove stale docs. Useful for the remove function which shouldn't be impacted by expirations
* @param {Function} callback Signature err, candidates
*/
async getCandidates (query, dontExpireStaleDocs, callback) {
const validDocs = []
getCandidates (query, dontExpireStaleDocs, callback) {
if (typeof dontExpireStaleDocs === 'function') {
callback = dontExpireStaleDocs
dontExpireStaleDocs = false
}
try {
callbackify(this.getCandidatesAsync.bind(this))(query, dontExpireStaleDocs, callback)
}
async getCandidatesAsync (query, dontExpireStaleDocs = false) {
const indexNames = Object.keys(this.indexes)
const validDocs = []
const _getCandidates = query => {
// STEP 1: get candidates list by checking indexes from most to least frequent usecase
const docs = this._getCandidates(query)
// For a basic match
let usableQuery
usableQuery = Object.entries(query)
.filter(([k, v]) =>
!!(typeof v === 'string' || typeof v === 'number' || typeof v === 'boolean' || isDate(v) || v === null) &&
indexNames.includes(k)
)
.pop()
if (usableQuery) return this.indexes[usableQuery[0]].getMatching(usableQuery[1])
// For a $in match
usableQuery = Object.entries(query)
.filter(([k, v]) =>
!!(query[k] && Object.prototype.hasOwnProperty.call(query[k], '$in')) &&
indexNames.includes(k)
)
.pop()
if (usableQuery) return this.indexes[usableQuery[0]].getMatching(usableQuery[1].$in)
// For a comparison match
usableQuery = Object.entries(query)
.filter(([k, v]) =>
!!(query[k] && (Object.prototype.hasOwnProperty.call(query[k], '$lt') || Object.prototype.hasOwnProperty.call(query[k], '$lte') || Object.prototype.hasOwnProperty.call(query[k], '$gt') || Object.prototype.hasOwnProperty.call(query[k], '$gte'))) &&
indexNames.includes(k)
)
.pop()
if (usableQuery) return this.indexes[usableQuery[0]].getBetweenBounds(usableQuery[1])
// By default, return all the DB data
return this.getAllData()
}
// STEP 1: get candidates list by checking indexes from most to least frequent usecase
const docs = _getCandidates(query)
// STEP 2: remove all expired documents
if (!dontExpireStaleDocs) {
const expiredDocsIds = []
@ -296,12 +332,7 @@ class Datastore extends EventEmitter {
})
}
} else validDocs.push(...docs)
} catch (error) {
if (typeof callback === 'function') callback(error, null)
else throw error
}
if (typeof callback === 'function') callback(null, validDocs)
else return validDocs
return validDocs
}
/**
@ -509,19 +540,30 @@ class Datastore extends EventEmitter {
options = {}
}
const callback = cb || (() => {})
const _callback = (err, res = {}) => {
callback(err, res.numAffected, res.affectedDocuments, res.upsert)
}
callbackify(this._updateAsync.bind(this))(query, updateQuery, options, _callback)
}
async _updateAsync (query, updateQuery, options = {}) {
const multi = options.multi !== undefined ? options.multi : false
const upsert = options.upsert !== undefined ? options.upsert : false
async.waterfall([
cb => { // If upsert option is set, check whether we need to insert the doc
if (!upsert) return cb()
// If upsert option is set, check whether we need to insert the doc
if (upsert) {
// Need to use an internal function not tied to the executor to avoid deadlock
const cursor = new Cursor(this, query)
const docs = await new Promise((resolve, reject) => {
cursor.limit(1)._exec((err, docs) => {
if (err) return callback(err)
if (docs.length === 1) return cb()
else {
if (err) reject(err)
else resolve(docs)
})
})
if (docs.length !== 1) {
let toBeInserted
try {
@ -531,32 +573,26 @@ class Datastore extends EventEmitter {
} catch (e) {
// updateQuery contains modifiers, use the find query as the base,
// strip it from all operators and update it according to updateQuery
try {
toBeInserted = model.modify(model.deepCopy(query, true), updateQuery)
} catch (err) {
return callback(err)
}
}
return this._insert(toBeInserted, (err, newDoc) => {
if (err) return callback(err)
return callback(null, 1, newDoc, true)
return new Promise((resolve, reject) => {
this._insert(toBeInserted, (err, newDoc) => {
if (err) return reject(err)
return resolve({ numAffected: 1, affectedDocuments: newDoc, upsert: true })
})
}
})
},
() => { // Perform the update
}
}
// Perform the update
let numReplaced = 0
let modifiedDoc
const modifications = []
let createdAt
this.getCandidates(query, (err, candidates) => {
if (err) return callback(err)
const candidates = await this.getCandidatesAsync(query)
// Preparing update (if an error is thrown here neither the datafile nor
// the in-memory indexes are affected)
try {
for (const candidate of candidates) {
if (model.match(candidate, query) && (multi || numReplaced === 0)) {
numReplaced += 1
@ -569,32 +605,25 @@ class Datastore extends EventEmitter {
modifications.push({ oldDoc: candidate, newDoc: modifiedDoc })
}
}
} catch (err) {
return callback(err)
}
// Change the docs in memory
try {
this.updateIndexes(modifications)
} catch (err) {
return callback(err)
}
// Update the datafile
const updatedDocs = modifications.map(x => x.newDoc)
await new Promise((resolve, reject) => {
this.persistence.persistNewState(updatedDocs, err => {
if (err) return callback(err)
if (!options.returnUpdatedDocs) {
return callback(null, numReplaced)
} else {
if (err) return reject(err)
else resolve()
})
})
if (!options.returnUpdatedDocs) return { numAffected: numReplaced }
else {
let updatedDocsDC = []
updatedDocs.forEach(doc => { updatedDocsDC.push(model.deepCopy(doc)) })
if (!multi) updatedDocsDC = updatedDocsDC[0]
return callback(null, numReplaced, updatedDocsDC)
return { numAffected: numReplaced, affectedDocuments: updatedDocsDC }
}
})
})
}])
}
update () {

Loading…
Cancel
Save