From: Ardis Lu <ardisglu@gmail.com>
Date: Sun, 19 Jan 2025 06:55:36 +0000 (-0800)
Subject: Refactor createCSVReadableStream to CSVReader
X-Git-Tag: v0.1.0~116
X-Git-Url: https://git.ardis.lu/?a=commitdiff_plain;h=c94ffa2b335a94eb41ac7ae28df6a694ed1c143c;p=csvio

Refactor createCSVReadableStream to CSVReader
---

diff --git a/README.md b/README.md
index 17b2a20..d625dbf 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ Run:
 
 ```javascript
 import {
-  createCSVReadableStream,
+  CSVReader,
   CSVTransformer,
   createCSVWritableStream
 } from './src/index.js';
@@ -28,7 +28,7 @@ function timesTwo(row) {
   return [Number(row[0]) * 2, Number(row[1]) * 2];
 }
 
-await createCSVReadableStream('./data/example-in.csv')
+await new CSVReader('./data/example-in.csv')
   .pipeThrough(new CSVTransformer(timesTwo))
   .pipeTo(createCSVWritableStream('./data/example-out.csv'));
 ```
@@ -69,14 +69,14 @@ Streamed CSV provides the bare minimum functions to accomplish **this specific w
 
 The core API is contained in `src/core.js` and exposes three functions to read, transform, and write CSV files.
 
-### `createCSVReadableStream`
+### `CSVReader`
 
-Create a `ReadableStream` from a local CSV file where each CSV row is one chunk.
+A `ReadableStream` where each chunk is one row from a local CSV file.
 
 ```javascript
-import { createCSVReadableStream } from './src/index.js';
+import { CSVReader } from './src/index.js';
 
-const readableStream = createCSVReadableStream('./data/example-in.csv');
+const readableStream = new CSVReader('./data/example-in.csv');
 ```
 
 ### `CSVTransformer`
diff --git a/examples/ex1.js b/examples/ex1.js
index 13f212c..7af1e20 100644
--- a/examples/ex1.js
+++ b/examples/ex1.js
@@ -8,12 +8,12 @@
 // - Each call to the transformation function passes 1 row of the input CSV.
 // - The expected output of the transformation function is another array representing 1 row of the output CSV.
 
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
 
 function timesTwo(row) {
   return [Number(row[0]) * 2, Number(row[1]) * 2];
 }
 
-await createCSVReadableStream(new URL('./data/ex1-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex1-in.csv', import.meta.url))
   .pipeThrough(new CSVTransformer(timesTwo))
   .pipeTo(createCSVWritableStream(new URL('./data/ex1-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/examples/ex2.js b/examples/ex2.js
index ac435b7..25f4c20 100644
--- a/examples/ex2.js
+++ b/examples/ex2.js
@@ -7,7 +7,7 @@
 // - If your input CSV does not have a header row, set `includeHeaders: true` and process the header row (first row) normally.
 // - Otherwise, you will need distinct logic to handle the header row vs. data rows. A simple boolean variable is a good way to switch between this logic. 
 
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
 
 let firstRow = true;
 function sum(row) {
@@ -18,6 +18,6 @@ function sum(row) {
   return [...row, Number(row[0]) + Number(row[1])];
 }
 
-await createCSVReadableStream(new URL('./data/ex2-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex2-in.csv', import.meta.url))
   .pipeThrough(new CSVTransformer(sum, { includeHeaders: true }))
   .pipeTo(createCSVWritableStream(new URL('./data/ex2-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/examples/ex3.js b/examples/ex3.js
index 95c0f28..8e44dc3 100644
--- a/examples/ex3.js
+++ b/examples/ex3.js
@@ -5,7 +5,7 @@
 // transformations (e.g., transformations spread across multiple functions or lower-level transformations directly
 // on byte arrays).
 
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
 
 // Parse the raw CSV data and transform the data into an object for convenience
 function parse(row) {
@@ -22,7 +22,7 @@ function calculate(obj) {
   return [name, paddedId];
 }
 
-await createCSVReadableStream(new URL('./data/ex3-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex3-in.csv', import.meta.url))
   .pipeThrough(new CSVTransformer(parse, { includeHeaders: true, rawOutput: true }))
   .pipeThrough(new CSVTransformer(calculate, { includeHeaders: true, rawInput: true }))
   .pipeTo(createCSVWritableStream(new URL('./data/ex3-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/examples/ex4.js b/examples/ex4.js
index 9039354..59695c9 100644
--- a/examples/ex4.js
+++ b/examples/ex4.js
@@ -8,7 +8,7 @@
 // `createCSVNormalizationStream` performs transformations to un-mangle the above using only a declaration of the header names and data types.
 // `createCSVDenormalizationStream` converts the data stream back into a CSV row.
 
-import { createCSVReadableStream, createCSVWritableStream, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/index.js';
+import { CSVReader, createCSVWritableStream, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/index.js';
 
 // The order of items in `headers` is significant, it determines the order of the output CSV columns.
 // If an input column is not provided in `headers`, the column is removed from the output CSV.
@@ -30,7 +30,7 @@ const headers = [
   }
 ];
 
-await createCSVReadableStream(new URL('./data/ex4-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex4-in.csv', import.meta.url))
   .pipeThrough(createCSVNormalizationStream(headers))
   .pipeThrough(createCSVDenormalizationStream())
   .pipeTo(createCSVWritableStream(new URL('./data/ex4-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/examples/ex5.js b/examples/ex5.js
index ab7c762..75130fb 100644
--- a/examples/ex5.js
+++ b/examples/ex5.js
@@ -1,11 +1,11 @@
 // ## Example 5: Handling an irregularly-shaped CSV file
 // 
-// `createCSVReadableStream` can parse CSVs that have rows with a mismatching number of columns, for example CSVs
+// `CSVReader` can parse CSVs that have rows with a mismatching number of columns, for example CSVs
 // with informational file headers. This information can be handled as required in the transformation function.
 //
 // Return `null` in the transformation function to consume an input row without emitting an output row.
 
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
 
 let i = 0;
 let report, id, date;
@@ -24,6 +24,6 @@ function parse(row) {
   return [report, id, date, ...row];
 }
 
-await createCSVReadableStream(new URL('./data/ex5-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex5-in.csv', import.meta.url))
   .pipeThrough(new CSVTransformer(parse, { includeHeaders: true }))
   .pipeTo(createCSVWritableStream(new URL('./data/ex5-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/examples/ex6.js b/examples/ex6.js
index 6f77a81..31aca2f 100644
--- a/examples/ex6.js
+++ b/examples/ex6.js
@@ -6,7 +6,7 @@
 // The `onError` function may be used to implement graceful failure strategies (e.g., set the output row to a
 // placeholder), retry strategies (e.g., a truncated exponential backoff for flaky network requests), logging, etc.
 
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
 
 // Will immediately retry calling `fn` a maximum of `iterations` times
 function retry(row, fn, iterations) {
@@ -25,6 +25,6 @@ function flaky(row) {
   return [`${v}: pass`];
 }
 
-await createCSVReadableStream(new URL('./data/ex6-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex6-in.csv', import.meta.url))
   .pipeThrough(new CSVTransformer(flaky, { onError: (row, e, fn) => retry(row, fn, 1000) }))
   .pipeTo(createCSVWritableStream(new URL('./data/ex6-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/examples/ex7.js b/examples/ex7.js
index f70c376..3f2b52a 100644
--- a/examples/ex7.js
+++ b/examples/ex7.js
@@ -6,7 +6,7 @@
 // This functionality is useful if you need to collect data from multiple rows before processing (e.g., performing an
 // expensive network request where it is more efficient to batch multiple rows into one request).
 
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
 
 let firstChunk = true;
 let batchNumber = 0;
@@ -28,6 +28,6 @@ function process(batch) {
   return batch.map(row => [batchNumber, sum, ...row]);
 }
 
-await createCSVReadableStream(new URL('./data/ex7-in.csv', import.meta.url))
+await new CSVReader(new URL('./data/ex7-in.csv', import.meta.url))
   .pipeThrough(new CSVTransformer(process, { includeHeaders: true, maxBatchSize: 5 }))
   .pipeTo(createCSVWritableStream(new URL('./data/ex7-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/src/core.js b/src/core.js
index d5a9d97..da558b6 100644
--- a/src/core.js
+++ b/src/core.js
@@ -55,76 +55,96 @@ export function parsePathLike(path) {
  * Uses Node.js's [`fs.createReadStream()`](https://nodejs.org/api/fs.html#fscreatereadstreampath-options) to read a local
  * CSV file then parses the file contents and returns the data through a [Streams API](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API)
  * `ReadableStream`.
- * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file.
- * @returns {ReadableStream<string>} A `ReadableStream` where each chunk is one row of the CSV file, in the form of a string
- * that may be deserialized using `JSON.parse()`.
+ * 
+ * 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 function createCSVReadableStream(path) {
-  const reader = createReadStream(path, { encoding: 'utf-8' });
-  const stream = Readable.toWeb(reader);
+export class CSVReader extends ReadableStream {
+  #row = [];
+  #field = '';
+  #lastChar = null;
+  #escapeMode = false;
+  #justExitedEscapeMode = false;
 
-  const row = [];
-  let field = '';
-  let lastChar = null;
-  let escapeMode = false;
-  let justExitedEscapeMode = false;
-  return stream.pipeThrough(new TransformStream({
-    transform(chunk, controller) {
-      for (const char of chunk) {
-        /* Escape mode logic */
-        if (escapeMode) {
-          if (char === '"') { // Exit escape mode and do not add char to field
-            escapeMode = false;
-            justExitedEscapeMode = true;
-          }
-          else { // Add the literal char because in escape mode
-            justExitedEscapeMode = false;
-            field += char;
-          }
+  /**
+   * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file.
+   */
+  constructor(path) {
+    const readStream = createReadStream(path, { encoding: 'utf-8' });
+    const readableStream = Readable.toWeb(readStream).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();
         }
-        /* Normal mode logic */
         else {
-          if (char === '"') {
-            if (justExitedEscapeMode) { // Re-enter escape mode because the exit char was escaped, and add char to field
-              escapeMode = true;
-              field += char;
-            }
-            else if (lastChar === ',' || lastChar === '\n' || lastChar === null) { // Enter escape mode and do not add char to field
-              escapeMode = true;
-            }
-            else { // No special cases match, add the literal char
-              field += char;
-            }
-          }
-          else if (char === '\r' || char === '\uFEFF') { } // Ignore CR (newline logic is handled by LF) and byte order mark
-          else if (char === ',') { // Terminate the field and do not add char to field
-            row.push(field);
-            field = '';
+          controller.enqueue(value);
+        }
+      }
+    });
+  }
+
+  #transform(chunk, controller) {
+    for (const char of chunk) {
+      /* Escape mode logic */
+      if (this.#escapeMode) {
+        if (char === '"') { // Exit escape mode and do not add char to field
+          this.#escapeMode = false;
+          this.#justExitedEscapeMode = true;
+        }
+        else { // Add the literal char because in escape mode
+          this.#justExitedEscapeMode = false;
+          this.#field += char;
+        }
+      }
+      /* Normal mode logic */
+      else {
+        if (char === '"') {
+          if (this.#justExitedEscapeMode) { // Re-enter escape mode because the exit char was escaped, and add char to field
+            this.#escapeMode = true;
+            this.#field += char;
           }
-          else if (char === '\n') { // Terminate the field and row and do not add char to field
-            row.push(field);
-            field = '';
-            controller.enqueue(JSON.stringify(row));
-            row.length = 0;
+          else if (this.#lastChar === ',' || this.#lastChar === '\n' || this.#lastChar === null) { // Enter escape mode and do not add char to field
+            this.#escapeMode = true;
           }
           else { // No special cases match, add the literal char
-            field += char;
+            this.#field += char;
           }
-          justExitedEscapeMode = false;
         }
-        lastChar = char;
-      }
-    },
-    flush(controller) {
-      if (field !== '' || row.length !== 0) { // CSV terminated without a trailing CRLF, leaving a row in the queue
-        row.push(field); // Last field is always un-pushed if CSV terminated with nothing
-        controller.enqueue(JSON.stringify(row));
-      }
-      else if (lastChar === null) { // CSV is blank
-        controller.enqueue(JSON.stringify(['']));
+        else if (char === '\r' || char === '\uFEFF') { } // Ignore CR (newline logic is handled by LF) and byte order mark
+        else if (char === ',') { // Terminate the field and do not add char to field
+          this.#row.push(this.#field);
+          this.#field = '';
+        }
+        else if (char === '\n') { // Terminate the field and row and do not add char to field
+          this.#row.push(this.#field);
+          this.#field = '';
+          controller.enqueue(JSON.stringify(this.#row));
+          this.#row.length = 0;
+        }
+        else { // No special cases match, add the literal char
+          this.#field += char;
+        }
+        this.#justExitedEscapeMode = false;
       }
+      this.#lastChar = char;
+    }
+  }
+
+  #flush(controller) {
+    if (this.#field !== '' || this.#row.length !== 0) { // CSV terminated without a trailing CRLF, leaving a row in the queue
+      this.#row.push(this.#field); // Last field is always un-pushed if CSV terminated with nothing
+      controller.enqueue(JSON.stringify(this.#row));
+    }
+    else if (this.#lastChar === null) { // CSV is blank
+      controller.enqueue(JSON.stringify(['']));
     }
-  }));
+  }
 }
 
 /**
@@ -138,7 +158,7 @@ export function createCSVReadableStream(path) {
  */
 
 /**
- * A function to process a row or rows of CSV data from `createCSVReadableStream`.
+ * A function to process a row or rows of CSV data from `CSVReader`.
  * @callback TransformationFunction
  * @param {TransformationInput} row A row or rows of input CSV data before transformation.
  * @returns {TransformationOutput} A row or rows of output CSV data after transformation, or null to skip a row.
diff --git a/src/index.js b/src/index.js
index 51ebea8..c75b66b 100644
--- a/src/index.js
+++ b/src/index.js
@@ -1,2 +1,2 @@
-export { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from './core.js';
+export { CSVReader, CSVTransformer, createCSVWritableStream } from './core.js';
 export { createCSVNormalizationStream, createCSVDenormalizationStream } from './normalization.js';
diff --git a/test/core.test.js b/test/core.test.js
index 9b08754..e850c27 100644
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -5,7 +5,7 @@ import { normalize, resolve } from 'node:path';
 import { pathToFileURL } from 'node:url';
 
 import { csvStreamEqualWritable, csvStreamNotEqualWritable, createCSVMockStream, createTempFile } from './utils.js';
-import { arrayToCSVString, parsePathLike, createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/core.js';
+import { arrayToCSVString, parsePathLike, CSVReader, CSVTransformer, createCSVWritableStream } from '../src/core.js';
 
 suite('arrayToCSVString', { concurrency: true }, () => {
   const vectors = [
@@ -52,7 +52,7 @@ suite('parsePathLike', { concurrency: true }, () => {
   }
 });
 
-suite('createCSVReadableStream', { concurrency: true }, () => {
+suite('CSVReader', { concurrency: true }, () => {
   const vectors = [
     {
       name: 'parses simple CSV',
@@ -183,9 +183,9 @@ suite('createCSVReadableStream', { concurrency: true }, () => {
   ];
   for (const { name, input, output, negativeOutput } of vectors) {
     test(name, { concurrency: true }, async () => {
-      await createCSVReadableStream(input).pipeTo(csvStreamEqualWritable(output));
+      await new CSVReader(input).pipeTo(csvStreamEqualWritable(output));
       if (negativeOutput !== undefined) {
-        await createCSVReadableStream(input).pipeTo(csvStreamNotEqualWritable(negativeOutput));
+        await new CSVReader(input).pipeTo(csvStreamNotEqualWritable(negativeOutput));
       }
     });
   }
@@ -422,7 +422,7 @@ suite('createCSVWritableStream', { concurrency: true }, () => {
       const temp = await createTempFile();
       t.after(async () => await unlink(temp));
       await createCSVMockStream(csv).pipeTo(createCSVWritableStream(temp));
-      await createCSVReadableStream(temp).pipeTo(csvStreamEqualWritable(csv));
+      await new CSVReader(temp).pipeTo(csvStreamEqualWritable(csv));
     });
   }
 });
\ No newline at end of file
diff --git a/test/normalization.test.js b/test/normalization.test.js
index 7f3e611..cf9c95a 100644
--- a/test/normalization.test.js
+++ b/test/normalization.test.js
@@ -2,7 +2,7 @@ import { suite, test } from 'node:test';
 import { deepStrictEqual } from 'node:assert/strict';
 
 import { csvStreamEqual, csvStreamEqualWritable, createCSVMockStream } from './utils.js';
-import { createCSVReadableStream } from '../src/core.js';
+import { CSVReader } from '../src/core.js';
 import { toCamelCase, expandScientificNotation, fixExcelNumber, fixExcelBigInt, fixExcelDate, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/normalization.js';
 
 suite('toCamelCase', { concurrency: true }, () => {
@@ -260,7 +260,7 @@ suite('createCSVNormalizationStream and createCSVDenormalizationStream end-to-en
       { name: 'bigintCol', displayName: 'BigInt Column', type: 'bigint' },
       { name: 'dateCol', displayName: 'Date Column', type: 'date' }
     ]
-    const stream = createCSVReadableStream('./test/data/normalization.csv')
+    const stream = new CSVReader('./test/data/normalization.csv')
       .pipeThrough(createCSVNormalizationStream(headers))
       .pipeThrough(createCSVDenormalizationStream());
     await csvStreamEqual(stream, [