Merge 6180ccd into 92cfdee

lib/database.js

@@ -62,44 +62,46 @@ Database.prototype.model = function(name, schema) {
  * @return {Promise}
  */
 Database.prototype.load = function(callback) {
-  const path = this.options.path;
+  const { path, onUpgrade, onDowngrade, version: newVersion } = this.options;
 
   if (!path) throw new WarehouseError('options.path is required');
 
-  return new Promise((resolve, reject) => {
-    const src = fs.createReadStream(path, {encoding: 'utf8'});
-    let oldVersion = 0;
-
-    const stream = JSONStream.parse([true, true], (value, keys) => {
-      switch (keys.shift()) {
-        case 'meta':
-          if (keys.shift() === 'version') {
-            oldVersion = value;
-          }
-
-          break;
-
-        case 'models':
-          this.model(keys.shift())._import(value);
-          break;
-      }
-    });
-
-    src
-      .pipe(stream)
-      .on('error', reject)
-      .on('end', () => {
-        resolve(oldVersion);
-      });
-  }).then(oldVersion => {
-    const newVersion = this.options.version;
+  let oldVersion = 0;
+
+  const getMetaCallBack = data => {
+    if (data.meta && data.meta.version) {
+      oldVersion = data.meta.version;
+    }
+  };
+
+  // data event arg0 wrap key/value pair.
+  const parseStream = JSONStream.parse('models.$*');
+
+  parseStream.once('header', getMetaCallBack);
+  parseStream.once('footer', getMetaCallBack);
 
+  parseStream.on('data', data => {
+    this.model(data.key)._import(data.value);
+  });
+
+  const rs = fs.createReadStream(path, 'utf8');
+
+  const promise = new Promise((resolve, reject) => {
+    parseStream.once('error', reject);
+    parseStream.once('end', resolve);
+
+    rs.once('error', reject);
+  }).then(() => {
     if (newVersion > oldVersion) {
-      return this.options.onUpgrade(oldVersion, newVersion);
+      return onUpgrade(oldVersion, newVersion);
     } else if (newVersion < oldVersion) {
-      return this.options.onDowngrade(oldVersion, newVersion);
+      return onDowngrade(oldVersion, newVersion);
     }
   }).asCallback(callback);
+
+  rs.pipe(parseStream);
+
+  return promise;
 };
 
 /**