From 27a612b33d0f2a67cf1d7dd3de3c04b8cb6e3e54 Mon Sep 17 00:00:00 2001 From: Ardis Lu Date: Sat, 16 Aug 2025 23:40:55 -0700 Subject: [PATCH] Add initial `TransformStream` CSVWriter implementation This change enables users to create a `TransformStream` by doing `new CSVWriter()` instead of always creating an ultimate sink `WritableStream`. This behavior is useful when there is some transformation that must be performed on the CSV data before writing (e.g., compressing into a `.csv.gz` file using `CompressionStream`). --- src/core.js | 125 +++++++++++++++++++++++++--------------- test/core.test.js | 38 +++++++++++- test/data/simple.csv.gz | Bin 82 -> 75 bytes test/utils.js | 31 +++++++++- 4 files changed, 145 insertions(+), 49 deletions(-) diff --git a/src/core.js b/src/core.js index 68eb006..5d35eee 100644 --- a/src/core.js +++ b/src/core.js @@ -437,7 +437,8 @@ export class CSVTransformer extends TransformStream { /** * @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 @@ -445,73 +446,107 @@ export class CSVTransformer extends TransformStream { * @property {boolean} done If `true`, the writer is finished writing to the output CSV file and the file handle is closed. */ +/** @type {WeakMap} 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} */ - 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,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; - } - 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(); - } - }, new CountQueuingStrategy({ highWaterMark: 4 })); // "4" picked based on manual benchmarks + } - this.#status = status; + /** @type {WritableStream|TransformStream,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 { + 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(); + } + }, new CountQueuingStrategy({ highWaterMark: 4 })); // "4" picked based on manual benchmarks + } + + // 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 } } diff --git a/test/core.test.js b/test/core.test.js index c4c8eb1..510b032 100644 --- a/test/core.test.js +++ b/test/core.test.js @@ -1,12 +1,12 @@ 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 }, () => { @@ -720,6 +720,32 @@ suite('CSVWriter', { 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 }, () => { @@ -730,6 +756,12 @@ 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)); diff --git a/test/data/simple.csv.gz b/test/data/simple.csv.gz index 33b18c193fa9a40a6bf647dc17839e37544c41c9..f13bea0caa63b5ab37ed4177b4e2b17c95e4b5aa 100644 GIT binary patch delta 53 zcmWIY7M1VjU;qOy?{nuh&z?2D;>mV-qP(o+r0Lp*hJosunx5*GmNTbwGi3eM*LDCZ F1ptnf4#xlh literal 82 zcmV-Y0ImNYiwFp1sB32e0CQ<=aBO8RV{>)@OU}>} + */ +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(); + } + }); +} -- 2.47.3