/**
* @typedef CSVWriterStatus The current status of the CSVWriter.
- * @property {string} name The name of the output CSV file that this writer is writing to.
+ * @property {string|null} name The name of the output CSV file that this writer is writing to, or `null` if the writer is
+ * a `TransformStream`.
* @property {number} elapsed The number of milliseconds that have elapsed between this writer's creation and the output CSV file
* handle closing, or until now if the writer is still writing. Measured using `performance.now()`.
* @property {number} rows The estimated number of CSV rows that have been written. If the data passed to the writer is
* @property {boolean} done If `true`, the writer is finished writing to the output CSV file and the file handle is closed.
*/
+/** @type {WeakMap<CSVWriter, CSVWriterStatus>} Simulated private field for CSVWriter, required to simulate a dynamic superclass. */
+const _status = new WeakMap();
+
/**
* Write a streamed CSV file to disk.
*
* A simple wrapper around Node.js's [`fs.writeFile`](https://nodejs.org/api/fs.html#fspromiseswritefilefile-data-options) to write
* streamed data to a file. If the input data is JSON, the data is converted to a string representing a RFC 4180 CSV record. If the
* data is not JSON (e.g., if it is already converted to a CSV string), the data is written to the CSV file directly.
- * @extends WritableStream
*/
-export class CSVWriter extends WritableStream {
- #status;
-
+export class CSVWriter {
/**
+ * @overload
* @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file destination. If the file already
* exists, its **data will be overwritten**.
+ * @returns {CSVWriter&WritableStream<any>}
*/
- constructor(path) {
- const fullPath = parsePathLike(path);
- const dir = dirname(fullPath);
+ /**
+ * @overload
+ * @param {null} [path] A `string`, `Buffer`, or `URL` representing a path to a local CSV file destination. If the file already
+ * exists, its **data will be overwritten**.
+ * @returns {CSVWriter&TransformStream<Array<string>,string>}
+ */
+ constructor(path = null) {
const status = {
- name: basename(fullPath),
start: performance.now(),
elapsed: 0,
rows: 0,
done: false
- };
- /** @type {FileHandle} */
- let handle;
- super({
- async start() {
- await mkdir(dir, { recursive: true });
- handle = await open(fullPath, 'w');
- },
- async write(chunk) {
- let data;
- if (Array.isArray(chunk)) {
- data = CSVWriter.arrayToCSVString(chunk);
- status.rows += chunk.length;
+ }
+
+ /** @type {WritableStream<any>|TransformStream<Array<string>,string>} */
+ let stream;
+ if (path === null) {
+ status.name = null;
+ stream = new TransformStream({
+ transform(chunk, controller) {
+ let data;
+ if (Array.isArray(chunk)) {
+ data = CSVWriter.arrayToCSVString(chunk);
+ status.rows += chunk.length;
+ }
+ else {
+ data = chunk;
+ status.rows++;
+ }
+ controller.enqueue(data);
}
- else {
- data = chunk;
- status.rows++;
+ });
+ }
+ else {
+ const fullPath = parsePathLike(path);
+ const dir = dirname(fullPath);
+ status.name = basename(fullPath);
+ /** @type {FileHandle} */
+ let handle;
+ stream = new WritableStream({
+ async start() {
+ await mkdir(dir, { recursive: true });
+ handle = await open(fullPath, 'w');
+ },
+ async write(chunk) {
+ let data;
+ if (Array.isArray(chunk)) {
+ data = CSVWriter.arrayToCSVString(chunk);
+ status.rows += chunk.length;
+ }
+ else {
+ data = chunk;
+ status.rows++;
+ }
+ await handle.write(data);
+ },
+ async close() {
+ await handle.close();
+ status.elapsed = performance.now() - status.start;
+ status.done = true;
+ },
+ async abort() {
+ await this.close();
}
- await handle.write(data);
- },
- async close() {
- await handle.close();
- status.elapsed = performance.now() - status.start;
- status.done = true;
- },
- async abort() {
- await this.close();
- }
- }, new CountQueuingStrategy({ highWaterMark: 4 })); // "4" picked based on manual benchmarks
+ }, new CountQueuingStrategy({ highWaterMark: 4 })); // "4" picked based on manual benchmarks
+ }
- this.#status = status;
+ // Proxy is required to simulate a dynamic superclass
+ const proxy = createProxy(this, stream);
+ _status.set(proxy, status);
+ return proxy;
}
/** @type {CSVWriterStatus} */
get status() {
- if (!this.#status.done) {
- this.#status.elapsed = performance.now() - this.#status.start;
+ const status = _status.get(this);
+ if (!status.done) {
+ status.elapsed = performance.now() - status.start;
}
return {
- name: this.#status.name,
- elapsed: this.#status.elapsed,
- rows: this.#status.rows,
- done: this.#status.done
+ name: status.name,
+ elapsed: status.elapsed,
+ rows: status.rows,
+ done: status.done
}
}
import { suite, test } from 'node:test';
import { ok, deepStrictEqual, throws, rejects } from 'node:assert/strict';
-import { unlink } from 'node:fs/promises';
+import { unlink, readFile } from 'node:fs/promises';
import { normalize, basename, resolve } from 'node:path';
import { setTimeout } from 'node:timers/promises';
import { pathToFileURL } from 'node:url';
-import { ReadableStream, DecompressionStream, TextDecoderStream } from 'node:stream/web';
+import { ReadableStream, DecompressionStream, CompressionStream, TextDecoderStream } from 'node:stream/web';
-import { csvStreamEqualWritable, createCSVMockStream, createTempFile } from './utils.js';
+import { csvStreamEqualWritable, createCSVMockStream, createTempFile, createWritableFileStream } from './utils.js';
import { parsePathLike, createFileStream, CSVReader, CSVTransformer, CSVWriter } from '../src/core.js';
suite('CSVWriter.arrayToCSVString', { concurrency: true }, () => {
deepStrictEqual(s instanceof CSVWriter, true);
await s.abort();
});
+ test('can create TransformStream (raw strings)', { concurrency: true }, async (t) => {
+ const input = new URL('./data/simple.csv', import.meta.url);
+ const stream = new CSVReader(input).pipeThrough(new CSVWriter());
+ const expected = [
+ 'column1,column2,column3\r\n',
+ 'abc,def,ghi\r\n',
+ '123,456,789\r\n',
+ 'aaa,bbb,ccc\r\n'
+ ];
+ for await (const row of stream) {
+ deepStrictEqual(row, expected.shift());
+ }
+ });
+ test('can create TransformStream (simple.csv.gz)', { concurrency: true }, async (t) => {
+ const temp = await createTempFile();
+ t.after(async () => await unlink(temp));
+ const input = new URL('./data/simple.csv', import.meta.url);
+ const output = new URL('./data/simple.csv.gz', import.meta.url);
+ await new CSVReader(input)
+ .pipeThrough(new CSVWriter())
+ .pipeThrough(new CompressionStream('gzip'))
+ .pipeTo(createWritableFileStream(temp));
+ const expected = await readFile(output);
+ const actual = await readFile(temp);
+ ok(actual.equals(expected));
+ });
});
suite('CSVWriter status', { concurrency: true }, () => {
await createCSVMockStream([['']]).pipeTo(writer);
deepStrictEqual(writer.status.name, basename(temp));
});
+ test('records file name of "null" if CSVWriter is a TransformStream', { concurrency: true }, async (t) => {
+ const writer = new CSVWriter();
+ const stream = createCSVMockStream([['']]).pipeThrough(writer);
+ deepStrictEqual(writer.status.name, null);
+ for await (const _ of stream) { }
+ });
test('records elapsed time', { concurrency: true }, async (t) => {
const temp = await createTempFile();
t.after(async () => await unlink(temp));
import { normalize } from 'node:path';
import { ReadableStream } from 'node:stream/web';
/** @import { TestContext } from 'node:test'; */
+/** @import { PathLike } from 'node:fs'; */
+/** @import { FileHandle } from 'node:fs/promises'; */
import { utils } from '../src/errorStrategies.js';
}
})();
return ReadableStream.from(chunks);
-}
\ No newline at end of file
+}
+
+/**
+ * Creates a writable file stream for a local file.
+ *
+ * @see {@link https://streams.spec.whatwg.org/#example-ws-backpressure}
+ * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local file.
+ * @returns {WritableStream<Uint8Array<ArrayBuffer>>}
+ */
+export function createWritableFileStream(path) {
+ /** @type {FileHandle} */
+ let handle;
+
+ return new WritableStream({
+ async start() {
+ handle = await open(path, 'w');
+ },
+ async write(chunk) {
+ await handle.write(chunk, 0, chunk.length);
+ },
+ async close() {
+ await handle.close();
+ },
+ async abort() {
+ await handle.close();
+ }
+ });
+}
git.ardis.lu / csvio / commitdiff