WIP: remove async from persistence module

Timothée Rebours 3 years ago
parent 3a9401cb21
commit 161c2da7d7
  1. 120
      lib/persistence.js
  2. 6
      lib/storage.js

@ -5,7 +5,7 @@
* * Persistence.persistNewState(newDocs, callback) where newDocs is an array of documents and callback has signature err
*/
const path = require('path')
const async = require('async')
const { callbackify, promisify } = require('util')
const byline = require('./byline')
const customUtils = require('./customUtils.js')
const Index = require('./indexes.js')
@ -73,9 +73,13 @@ class Persistence {
* @param {Function} callback Optional callback, signature: err
*/
persistCachedDatabase (callback = () => {}) {
return callbackify(this.persistCachedDatabaseAsync.bind(this))(callback)
}
async persistCachedDatabaseAsync () {
const lines = []
if (this.inMemoryOnly) return callback(null)
if (this.inMemoryOnly) return
this.db.getAllData().forEach(doc => {
lines.push(this.afterSerialization(model.serialize(doc)))
@ -92,11 +96,8 @@ class Persistence {
}
})
storage.crashSafeWriteFileLines(this.filename, lines, err => {
if (err) return callback(err)
this.db.emit('compaction.done')
return callback(null)
})
await storage.crashSafeWriteFileLinesAsync(this.filename, lines)
this.db.emit('compaction.done')
}
/**
@ -135,18 +136,22 @@ class Persistence {
* @param {Function} callback Optional, signature: err
*/
persistNewState (newDocs, callback = () => {}) {
callbackify(this.persistNewStateAsync.bind(this))(newDocs, err => callback(err))
}
async persistNewStateAsync (newDocs) {
let toPersist = ''
// In-memory only datastore
if (this.inMemoryOnly) return callback(null)
if (this.inMemoryOnly) return
newDocs.forEach(doc => {
toPersist += this.afterSerialization(model.serialize(doc)) + '\n'
})
if (toPersist.length === 0) return callback(null)
if (toPersist.length === 0) return
storage.appendFile(this.filename, toPersist, 'utf8', err => callback(err))
await storage.appendFileAsync(this.filename, toPersist, 'utf8')
}
/**
@ -232,6 +237,10 @@ class Persistence {
})
}
async treatRawStreamAsync (rawStream) {
return promisify(this.treatRawStream.bind(this))(rawStream)
}
/**
* Load the database
* 1) Create all indexes
@ -243,65 +252,42 @@ class Persistence {
* @param {Function} callback Optional callback, signature: err
*/
loadDatabase (callback = () => {}) {
callbackify(this.loadDatabaseAsync.bind(this))(err => callback(err))
}
async loadDatabaseAsync () {
this.db.resetIndexes()
// In-memory only datastore
if (this.inMemoryOnly) return callback(null)
async.waterfall([
cb => {
// eslint-disable-next-line node/handle-callback-err
Persistence.ensureDirectoryExists(path.dirname(this.filename), err => {
// TODO: handle error
// eslint-disable-next-line node/handle-callback-err
storage.ensureDatafileIntegrity(this.filename, err => {
// TODO: handle error
const treatedDataCallback = (err, treatedData) => {
if (err) return cb(err)
// Recreate all indexes in the datafile
Object.keys(treatedData.indexes).forEach(key => {
this.db.indexes[key] = new Index(treatedData.indexes[key])
})
// Fill cached database (i.e. all indexes) with data
try {
this.db.resetIndexes(treatedData.data)
} catch (e) {
this.db.resetIndexes() // Rollback any index which didn't fail
return cb(e)
}
this.db.persistence.persistCachedDatabase(cb)
}
if (storage.readFileStream) {
// Server side
const fileStream = storage.readFileStream(this.filename, { encoding: 'utf8' })
this.treatRawStream(fileStream, treatedDataCallback)
return
}
// Browser
storage.readFile(this.filename, 'utf8', (err, rawData) => {
if (err) return cb(err)
try {
const treatedData = this.treatRawData(rawData)
treatedDataCallback(null, treatedData)
} catch (e) {
return cb(e)
}
})
})
})
}
], err => {
if (err) return callback(err)
this.db.executor.processBuffer()
return callback(null)
if (this.inMemoryOnly) return
await Persistence.ensureDirectoryExistsAsync(path.dirname(this.filename)) // TODO: maybe ignore error
await storage.ensureDatafileIntegrityAsync(this.filename) // TODO: maybe ignore error
let treatedData
if (storage.readFileStream) {
// Server side
const fileStream = storage.readFileStream(this.filename, { encoding: 'utf8' })
treatedData = await this.treatRawStreamAsync(fileStream)
} else {
// Browser
const rawData = await storage.readFileAsync(this.filename, 'utf8')
treatedData = this.treatRawData(rawData)
}
// Recreate all indexes in the datafile
Object.keys(treatedData.indexes).forEach(key => {
this.db.indexes[key] = new Index(treatedData.indexes[key])
})
// Fill cached database (i.e. all indexes) with data
try {
this.db.resetIndexes(treatedData.data)
} catch (e) {
this.db.resetIndexes() // Rollback any index which didn't fail
throw e
}
await this.db.persistence.persistCachedDatabaseAsync()
this.db.executor.processBuffer()
}
/**
@ -312,6 +298,10 @@ class Persistence {
storage.mkdir(dir, { recursive: true }, err => { callback(err) })
}
static async ensureDirectoryExistsAsync (dir) {
await storage.mkdirAsync(dir, { recursive: true })
}
/**
* Return the path the datafile if the given filename is relative to the directory where Node Webkit stores
* data for this application. Probably the best place to store data

@ -20,12 +20,16 @@ storage.rename = fs.rename
storage.renameAsync = fsPromises.rename
storage.writeFile = fs.writeFile
storage.writeFileAsync = fsPromises.writeFile
storage.writeFileStream = fs.createWriteStream
storage.unlink = fs.unlink
storage.unlinkAsync = fsPromises.unlink
storage.appendFile = fs.appendFile
storage.appendFileAsync = fsPromises.appendFile
storage.readFile = fs.readFile
storage.readFileAsync = fsPromises.readFile
storage.readFileStream = fs.createReadStream
storage.mkdir = fs.mkdir
storage.mkdirAsync = fsPromises.mkdir
/**
* Explicit name ...
@ -90,7 +94,7 @@ storage.flushToStorageAsync = async (options) => {
*/
storage.writeFileLines = (filename, lines, callback = () => {}) => {
try {
const stream = fs.createWriteStream(filename)
const stream = storage.writeFileStream(filename)
const readable = Readable.from(lines)
readable.on('data', (line) => {
try {

Loading…
Cancel
Save