Bring Reader into conformance with other resumable decoders

index.d.ts

@@ -102,4 +102,10 @@ export class Reader {
    * Async iterator support
    */
   [Symbol.asyncIterator](): AsyncIterableIterator<string[]>;
+
+  /**
+   * Raw encoded text of the row in progress, as consumed so far
+   * @returns The unparsed text; empty when there is none
+   */
+  partial(): string;
 }

index.js

@@ -2,9 +2,9 @@
  * NSV (Newline-Separated Values) format parser and serializer
  *
  * NSV is a plain text data format for sequences of sequences.
- * - Single newlines separate cells within a row
- * - Double newlines separate rows
- * - Backslash escapes: \\ for \, \n for newline, \ for empty cell
+ * - Each cell is terminated by a newline
+ * - Each row is terminated by an additional newline (an empty line)
+ * - Backslash escapes: \\ for \, \n for newline, lone \ for empty cell
  */
 
 /**
@@ -16,6 +16,9 @@ function unescape(str) {
   if (str === '\\') {
     return '';
   }
+  if (!str.includes('\\')) {
+    return str;
+  }
 
   let result = '';
   let i = 0;
@@ -225,19 +228,19 @@ class Writer {
 
 /**
  * Create a reader for incrementally reading NSV rows
- * Truly streams data - parses rows as chunks arrive without buffering entire input
  */
 class Reader {
   /**
    * @param {NodeJS.ReadableStream|string} input - Stream or string to read from
    */
   constructor(input) {
     this.input = input;
-    this._buffer = '';
-    this._currentRow = [];
+    this._partial = '';
+    this._atLineStart = true;
     this._rowQueue = [];
-    this._done = false;
+    this._rowQueueHead = 0;
     this._started = false;
+    this._ended = false;
     this._error = null;
   }
 
@@ -252,7 +255,7 @@ class Reader {
     // If input is a string, process it directly
     if (typeof this.input === 'string') {
       this._processChunk(this.input);
-      this._finalize();
+      this._ended = true;
       return;
     }
 
@@ -267,7 +270,7 @@ class Reader {
     });
 
     this.input.on('end', () => {
-      this._finalize();
+      this._ended = true;
     });
 
     this.input.on('error', (error) => {
@@ -280,42 +283,24 @@ class Reader {
    * @private
    */
   _processChunk(text) {
-    for (let i = 0; i < text.length; i++) {
-      const char = text[i];
-
-      if (char === '\n') {
-        if (this._buffer.length === 0) {
-          // Empty line - row complete
-          this._rowQueue.push(this._currentRow);
-          this._currentRow = [];
-        } else {
-          // Content before this newline - it's a cell
-          this._currentRow.push(unescape(this._buffer));
-          this._buffer = '';
-        }
-      } else {
-        // Regular character
-        this._buffer += char;
-      }
+    if (text.length === 0) {
+      return;
     }
-  }
-
-  /**
-   * Finalize parsing when stream ends
-   * @private
-   */
-  _finalize() {
-    // Handle any remaining buffered content
-    if (this._buffer.length > 0) {
-      this._currentRow.push(unescape(this._buffer));
+    let end = text.lastIndexOf('\n\n');
+    if (end !== -1) {
+      end += 2;
+    } else if (text[0] === '\n' && this._atLineStart) {
+      end = 1;
     }
-
-    // Add final row if it has content
-    if (this._currentRow.length > 0) {
-      this._rowQueue.push(this._currentRow);
+    this._atLineStart = text[text.length - 1] === '\n';
+    if (end === -1) {
+      this._partial += text;
+      return;
     }
-
-    this._done = true;
+    for (const row of parse(this._partial + text.slice(0, end))) {
+      this._rowQueue.push(row);
+    }
+    this._partial = end < text.length ? text.slice(end) : '';
   }
 
   /**
@@ -326,16 +311,21 @@ class Reader {
     this._start();
 
     // Wait for a row to be available or stream to finish
-    while (this._rowQueue.length === 0 && !this._done) {
+    while (this._rowQueueHead === this._rowQueue.length && !this._ended) {
       if (this._error) throw this._error;
       // Wait a tick for more data
       await new Promise(resolve => setImmediate(resolve));
     }
 
     if (this._error) throw this._error;
 
-    if (this._rowQueue.length > 0) {
-      return this._rowQueue.shift();
+    if (this._rowQueueHead < this._rowQueue.length) {
+      const row = this._rowQueue[this._rowQueueHead++];
+      if (this._rowQueueHead === this._rowQueue.length) {
+        this._rowQueue = [];
+        this._rowQueueHead = 0;
+      }
+      return row;
     }
 
     return null;
@@ -346,8 +336,6 @@ class Reader {
    * @returns {Promise<string[][]>} All remaining rows
    */
   async readRows() {
-    this._start();
-
     const rows = [];
     let row;
     while ((row = await this.readRow()) !== null) {
@@ -360,13 +348,21 @@ class Reader {
    * Async iterator support
    */
   async *[Symbol.asyncIterator]() {
-    this._start();
-
     let row;
     while ((row = await this.readRow()) !== null) {
       yield row;
     }
   }
+
+  /**
+   * Raw encoded text of the row in progress, as consumed so far
+   * @returns {string} The unparsed text; empty when there is none
+   */
+  partial() {
+    this._start();
+
+    return this._partial;
+  }
 }
 
 /**

test.js

@@ -184,7 +184,7 @@ async function runTests() {
 
   // Test 20: Reader - readRow
   {
-    const reader = new nsv.Reader('a\nb\n\nc\nd\n');
+    const reader = new nsv.Reader('a\nb\n\nc\nd\n\n');
     const row1 = await reader.readRow();
     const row2 = await reader.readRow();
     const row3 = await reader.readRow();
@@ -196,15 +196,15 @@ async function runTests() {
 
   // Test 21: Reader - readRows
   {
-    const reader = new nsv.Reader('a\nb\n\nc\nd\n');
+    const reader = new nsv.Reader('a\nb\n\nc\nd\n\n');
     const rows = await reader.readRows();
     assertEqual(rows, [['a', 'b'], ['c', 'd']], 'Reader - readRows');
     console.log('✓ Reader - readRows');
   }
 
   // Test 22: Reader - async iterator
   {
-    const reader = new nsv.Reader('a\nb\n\nc\nd\n');
+    const reader = new nsv.Reader('a\nb\n\nc\nd\n\n');
     const rows = [];
     for await (const row of reader) {
       rows.push(row);
@@ -215,7 +215,7 @@ async function runTests() {
 
   // Test 23: Reader from stream
   {
-    const stream = createReadableStream('a\nb\n\nc\nd\n');
+    const stream = createReadableStream('a\nb\n\nc\nd\n\n');
     const reader = new nsv.Reader(stream);
     const rows = await reader.readRows();
     assertEqual(rows, [['a', 'b'], ['c', 'd']], 'Reader from stream');
@@ -237,6 +237,51 @@ async function runTests() {
     console.log('✓ Reader - empty rows match parse()');
   }
 
+  // Test 23c: Reader withholds an unterminated final row
+  {
+    const reader1 = new nsv.Reader('a\nb\n\nc\nd');
+    assertEqual(await reader1.readRow(), ['a', 'b'], 'Reader - row before unterminated row');
+    assertEqual(await reader1.readRow(), null, 'Reader - unterminated row not emitted');
+
+    const reader2 = new nsv.Reader('a\nb\n\nc\nd\n');
+    assertEqual(await reader2.readRow(), ['a', 'b'], 'Reader - row before cell-terminated row');
+    assertEqual(await reader2.readRow(), null, 'Reader - cell-terminated row not emitted');
+
+    assertEqual(nsv.parse('a\nb\n\nc\nd'), [['a', 'b'], ['c', 'd']], 'parse - emits unterminated row');
+    console.log('✓ Reader withholds an unterminated final row');
+  }
+
+  // Test 23d: partial() exposes the row in progress
+  {
+    const inputs = [
+      'a\nb\n\nc\nd',
+      'a\nb\n\nc\nd\n',
+      'a\nx\\',
+      '\\',
+      'a\nb\n\n',
+      '',
+      '\n',
+    ];
+    for (const input of inputs) {
+      const reader = new nsv.Reader(input);
+      const rows = await reader.readRows();
+      assertEqual(
+        rows.concat(nsv.parse(reader.partial())),
+        nsv.parse(input),
+        `partial - rows + parse(partial) == parse (${JSON.stringify(input)})`
+      );
+    }
+
+    const reader = new nsv.Reader('a\nb\n\nc\\nd');
+    await reader.readRows();
+    assertEqual(reader.partial(), 'c\\nd', 'partial - raw escaped text');
+
+    const reader2 = new nsv.Reader('a\nb\n\n');
+    await reader2.readRows();
+    assertEqual(reader2.partial(), '', 'partial - empty on terminated input');
+    console.log('✓ partial() exposes the row in progress');
+  }
+
   // Test 24: Empty data array
   {
     const result = nsv.stringify([]);