refactor(export): use createWriteStream to export json

package.json

@@ -14,7 +14,7 @@
     "build": "tsc -b",
     "clean": "tsc -b --clean",
     "eslint": "eslint src test",
-    "test": "mocha -r ts-node/register 'test/scripts/**/*.ts'",
+    "test": "mocha -r ts-node/register test/scripts/**/*.ts",
     "test-cov": "c8 --reporter=lcovonly --reporter=text-summary npm test",
     "typedoc": "typedoc --entryPointStrategy expand ./src"
   },

src/database.ts

@@ -1,61 +1,48 @@
 import { parse as createJsonParseStream } from './lib/jsonstream';
 import BluebirdPromise from 'bluebird';
-import { writev, promises as fsPromises, createReadStream } from 'graceful-fs';
+import { createReadStream, createWriteStream } from 'graceful-fs';
 import { pipeline, Stream } from 'stream';
 import Model from './model';
 import Schema from './schema';
 import SchemaType from './schematype';
 import WarehouseError from './error';
 import { logger } from 'hexo-log';
 import type { AddSchemaTypeOptions, NodeJSLikeCallback } from './types';
+import { asyncWriteToStream } from './util';
 
 const log = logger();
 const pkg = require('../package.json');
-const { open } = fsPromises;
 const pipelineAsync = BluebirdPromise.promisify(pipeline) as unknown as (...args: Stream[]) => BluebirdPromise<unknown>;
 
-let _writev: (handle: fsPromises.FileHandle, buffers: Buffer[]) => Promise<unknown>;
-
-if (typeof writev === 'function') {
-  _writev = (handle, buffers) => handle.writev(buffers);
-} else {
-  _writev = async (handle, buffers) => {
-    for (const buffer of buffers) await handle.write(buffer);
-  };
-}
-
 async function exportAsync(database: Database, path: string): Promise<void> {
-  const handle = await open(path, 'w');
+  const writeStream = createWriteStream(path, { flags: 'w' });
 
   try {
+    let p: Promise<unknown> | undefined;
     // Start body & Meta & Start models
-    await handle.write(`{"meta":${JSON.stringify({
+    p = asyncWriteToStream(writeStream, `{"meta":${JSON.stringify({
       version: database.options.version,
       warehouse: pkg.version
     })},"models":{`);
+    if (p) await p;
 
     const models = database._models;
     const keys = Object.keys(models);
     const { length } = keys;
-
     // models body
     for (let i = 0; i < length; i++) {
       const key = keys[i];
 
       if (!models[key]) continue;
 
-      const buffers = [];
-
-      if (i) buffers.push(Buffer.from(',', 'ascii'));
-
-      buffers.push(Buffer.from(`"${key}":`));
-
-      buffers.push(Buffer.from(models[key]._export()));
-      await _writev(handle, buffers);
+      const prefix = i ? ',' : '';
+      p = asyncWriteToStream(writeStream, `${prefix}"${key}":`);
+      if (p) await p;
+      await models[key].toJSONStream(writeStream);
     }
-
     // End models
-    await handle.write('}}');
+    p = asyncWriteToStream(writeStream, '}}');
+    if (p) await p;
   } catch (e) {
     log.error(e);
     if (e instanceof RangeError && e.message.includes('Invalid string length')) {
@@ -67,7 +54,9 @@ async function exportAsync(database: Database, path: string): Promise<void> {
       throw e;
     }
   } finally {
-    await handle.close();
+    await new Promise<void>((resolve, reject) => {
+      writeStream.end(resolve);
+    });
   }
 }
 

src/model.ts

@@ -2,7 +2,7 @@ import { EventEmitter } from 'events';
 import rfdc from 'rfdc';
 const cloneDeep = rfdc();
 import BluebirdPromise from 'bluebird';
-import { parseArgs, getProp, setGetter, shuffle } from './util';
+import { parseArgs, getProp, setGetter, shuffle, asyncWriteToStream } from './util';
 import Document from './document';
 import Query from './query';
 import Schema from './schema';
@@ -12,6 +12,7 @@ import PopulationError from './error/population';
 import Mutex from './mutex';
 import type Database from './database';
 import type { AddSchemaTypeOptions, NodeJSLikeCallback, Options, queryCallback } from './types';
+import type { Writable } from 'node:stream';
 
 class Model<T> extends EventEmitter {
   _mutex = new Mutex();
@@ -1010,6 +1011,26 @@ class Model<T> extends EventEmitter {
     return JSON.stringify(this.toJSON());
   }
 
+  async toJSONStream(writeStream: Writable): Promise<void> {
+    let p: Promise<unknown> | undefined;
+    const { data, schema } = this;
+    const keys = this.dataKeys;
+    const { length } = keys;
+
+    p = asyncWriteToStream(writeStream, '[');
+    if (p) await p;
+    for (let i = 0; i < length; i++) {
+      const raw = data[keys[i]];
+      if (raw) {
+        const prefix = i === 0 ? '' : ',';
+        p = asyncWriteToStream(writeStream, `${prefix}${JSON.stringify(schema._exportDatabase(cloneDeep(raw) as object))}`);
+        if (p) await p;
+      }
+    }
+    p = asyncWriteToStream(writeStream, ']');
+    if (p) await p;
+  }
+
   toJSON(): any[] {
     const result = new Array(this.length);
     const { data, schema } = this;

src/util.ts

@@ -1,3 +1,6 @@
+import { once } from 'node:events';
+import type { Writable } from 'node:stream';
+
 function extractPropKey(key: string): string[] {
   return key.split('.');
 }
@@ -191,3 +194,24 @@ export function parseArgs<B extends string | Record<string, number | Record<stri
 
   return result;
 }
+
+/**
+ * A small utility function for writing chunk to a stream, and only return promise when needed (stream backpressure)
+ *
+ * ```ts
+ * const writeStream = fs.createWriteStream(outputFile);
+ * for (const line of source) {
+ *   const p = asyncWriteToStream(writeStream, line + '\n');
+ *   if (p) {
+ *     // eslint-disable-next-line no-await-in-loop -- stream backpressure
+ *     await p;
+ *   }
+ * }
+ */
+export function asyncWriteToStream<T>(stream: Writable, chunk: T): Promise<unknown> | null {
+  const waitDrain = !stream.write(chunk);
+  if (waitDrain) {
+    return once(stream, 'drain');
+  }
+  return null;
+}