From 54f98d5b22a22e8a2ef7ad62ce670abcbcc29698 Mon Sep 17 00:00:00 2001 From: Ardis Lu Date: Tue, 29 Jul 2025 23:53:47 -0700 Subject: [PATCH] Add initial `TransformStream` CSVReader implementation This change enables users to create a `TransformStream` by doing `new CSVReader()` instead of always creating an underlying source `ReadableStream`. This behavior is useful when there is some transformation that must be performed on the source file before parsing (e.g., decompressing a `.csv.gz` file using `DecompressionStream`). --- src/core.js | 71 ++++++++++++++++++++++++++-------------- test/core.test.js | 15 ++++++++- test/data/simple.csv.gz | Bin 0 -> 82 bytes 3 files changed, 61 insertions(+), 25 deletions(-) create mode 100644 test/data/simple.csv.gz diff --git a/src/core.js b/src/core.js index 8133fc5..d620724 100644 --- a/src/core.js +++ b/src/core.js @@ -1,7 +1,7 @@ import { mkdir, open } from 'node:fs/promises'; import { dirname, basename, normalize } from 'node:path'; import { fileURLToPath } from 'node:url'; -import { ReadableStream, WritableStream, TransformStream } from 'node:stream/web'; +import { ReadableStream, WritableStream, TransformStream, TextDecoderStream } from 'node:stream/web'; /** @import { PathLike } from 'node:fs'; */ /** @import { FileHandle } from 'node:fs/promises'; */ @@ -82,6 +82,26 @@ export function createFileStream(path) { }); } +/** + * Creates a simple `Proxy` that tries property access on `primary` and falls back to `underlying` if the property + * does not exist in `primary`. Only the `get` trap is implemented. + * + * @template T, U + * @param {T} primary The primary target object of the proxy. + * @param {U} underlying The fallback object to proxy property accesses to if not found on `primary`. + * @returns {T&U} A `Proxy` that behaves like `primary`, with fallback to `underlying` for missing properties. + */ +function createProxy(primary, underlying) { + return new Proxy(primary, { + get(target, prop, receiver) { + if (prop in target) { + return Reflect.get(target, prop, receiver); + } + return Reflect.get(underlying, prop, receiver); + } + }); +} + /** * Options to configure `CSVReader`. * @typedef {Object} CSVReaderOptions @@ -93,9 +113,8 @@ export function createFileStream(path) { * A simple streaming parser for CSV files. * * Each chunk is one row of the CSV file, in the form of a string that may be deserialized using `JSON.parse()`. - * @extends ReadableStream */ -export class CSVReader extends ReadableStream { +export class CSVReader { /** @type {Array} */ #row = []; #field = ''; @@ -105,29 +124,33 @@ export class CSVReader extends ReadableStream { #justExitedEscapeMode = false; /** - * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file. - * @param {CSVReaderOptions} options Object containing flags to configure the reader. + * @overload + * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file or + * `null` to create a `TransformStream`. + * @param {CSVReaderOptions} [options] Object containing flags to configure the reader. + * @returns {CSVReader&ReadableStream} */ - constructor(path, options = {}) { - const { encoding = 'utf-8' } = options; - const readableStream = createFileStream(path) - .pipeThrough(new TextDecoderStream(encoding)) // Byte order mark (BOM) is removed, if present - .pipeThrough(new TransformStream({ - transform: (chunk, controller) => this.#transform(chunk, controller), - flush: (controller) => this.#flush(controller) - })); - const reader = readableStream.getReader(); - super({ - async pull(controller) { - const { done, value } = await reader.read(); - if (done) { - controller.close(); - } - else { - controller.enqueue(value); - } - } + /** + * @overload + * @param {null} [path] A `string`, `Buffer`, or `URL` representing a path to a local CSV file or + * `null` to create a `TransformStream`. + * @returns {CSVReader&TransformStream} + */ + constructor(path = null, options = {}) { + /** @type {ReadableStream|TransformStream} */ + let stream; + stream = new TransformStream({ + transform: (chunk, controller) => this.#transform(chunk, controller), + flush: (controller) => this.#flush(controller) }); + if (path !== null) { + const { encoding = 'utf-8' } = options; + stream = createFileStream(path) + .pipeThrough(new TextDecoderStream(encoding)) // Byte order mark (BOM) is removed, if present + .pipeThrough(stream); + } + // Proxy is required to simulate a dynamic superclass + return createProxy(this, stream); } /** @type {import('node:stream/web').TransformerTransformCallback} */ diff --git a/test/core.test.js b/test/core.test.js index 7cde8e5..329eb3a 100644 --- a/test/core.test.js +++ b/test/core.test.js @@ -4,7 +4,7 @@ import { unlink } from 'node:fs/promises'; import { normalize, basename, resolve } from 'node:path'; import { setTimeout } from 'node:timers/promises'; import { pathToFileURL } from 'node:url'; -import { ReadableStream } from 'node:stream/web'; +import { ReadableStream, DecompressionStream, TextDecoderStream } from 'node:stream/web'; import { csvStreamEqualWritable, csvStreamNotEqualWritable, createCSVMockStream, createTempFile } from './utils.js'; import { parsePathLike, createFileStream, CSVReader, CSVTransformer, CSVWriter } from '../src/core.js'; @@ -299,6 +299,19 @@ suite('CSVReader', { concurrency: true }, () => { deepStrictEqual(s instanceof CSVReader, true); await s.cancel(); }); + test('can create TransformStream (simple.csv.gz)', { concurrency: true }, async () => { + const path = new URL('./data/simple.csv.gz', import.meta.url); + await createFileStream(path) + .pipeThrough(new DecompressionStream('gzip')) + .pipeThrough(new TextDecoderStream()) + .pipeThrough(new CSVReader()) + .pipeTo(csvStreamEqualWritable([ + ['column1', 'column2', 'column3'], + ['abc', 'def', 'ghi'], + ['123', '456', '789'], + ['aaa', 'bbb', 'ccc'] + ])); + }); }); suite('CSVTransformer', { concurrency: true }, () => { diff --git a/test/data/simple.csv.gz b/test/data/simple.csv.gz new file mode 100644 index 0000000000000000000000000000000000000000..33b18c193fa9a40a6bf647dc17839e37544c41c9 GIT binary patch literal 82 zcmV-Y0ImNYiwFp1sB32e0CQ<=aBO8RV{>)@OU}