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`).

diff --git a/src/core.js b/src/core.js
index 8133fc5911f95eb4153c1f677f48584625beebec..d620724f759b12456d541f8e8ef54c1532e29d51 100644 (file)
--- 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<string>
  */
-export class CSVReader extends ReadableStream {
+export class CSVReader {
   /** @type {Array<string>} */
   #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<string>}
    */
-  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<string,string>}
+   */
+  constructor(path = null, options = {}) {
+    /** @type {ReadableStream<string>|TransformStream<string,string>} */
+    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<string,string>} */

diff --git a/test/core.test.js b/test/core.test.js
index 7cde8e557d637d8ee2eb4a2af2dbf705b1c6ec9d..329eb3a8b5a0f6a1cf48b9a33e2d43c5020761dd 100644 (file)
--- 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 (file)
index 0000000..33b18c1
Binary files /dev/null and b/test/data/simple.csv.gz differ