Commit

Author: qwtel

json-parse-stream.ts

@@ -2,8 +2,6 @@
 import { streamToAsyncIter } from 'https://ghuc.cc/qwtel/whatwg-stream-to-async-iter/index.ts'
 import { JSONParser } from './json-parser.js';
 import { normalize, match } from './json-path.ts'
-// import { AsyncQueue } from './async-queue.ts';
-// import { BinarySplitStream } from './split-stream.ts'
 
 async function* _identity<T>(iter: Iterable<T> | AsyncIterable<T>) {
   for await (const x of iter) yield x;
@@ -16,64 +14,89 @@ const mkPath = (parser: any) => {
   return normalize(path.join('.')); // FIXME: avoid string concatenation/joining
 }
 
-/**
- * 
- */
 export class JSONParseStream<T = any> extends TransformStream<string | Uint8Array, T> {
-  #pathMap = new Map<any, string>(); // FIXME: clear when processing is done!?
   #jsonPath;
-  // #streams = new Map<string, ReadableStream<unknown>>();
 
   constructor(jsonPath = '$.*') {
     let parser!: JSONParser;
+    const expr = normalize(jsonPath)
     super({
       start: (controller) => {
         parser = new JSONParser();
         parser.onValue = (value: T) => {
           const path = mkPath(parser)
 
-          // FIXME: better solution?
-          this.#pathMap.set(value, path);
-          controller.enqueue(value);
+          if (match(expr, path)) {
+            controller.enqueue(value as any);
+          } else if (expr.startsWith(path)) {
+            // Closing the stream early when the selected path can no longer yield values.
+            controller.terminate()
+          }
         };
       },
       transform: (chunk) => {
         parser.write(chunk);
       },
     });
-    const expr = normalize(jsonPath)
     this.#jsonPath = expr;
   }
 
+  get path() { return this.#jsonPath }
+}
+
+export class JSONParseWritable<T = any> extends WritableStream<string | Uint8Array> {
+  #pathMap = new Map<any, string>(); // FIXME: clear when processing is done!?
+  #readable: ReadableStream<T>;
+  // #streams = new Map<string, ReadableStream<unknown>>();
+
+  constructor() {
+    let parser: JSONParser;
+    let readable: ReadableStream<T>
+    super({
+      start: (writeCtrl) => {
+        parser = new JSONParser();
+        readable = new ReadableStream({
+          start: (readCtrl) => {
+            parser.onValue = (value: T) => {
+              const path = mkPath(parser)
+
+              // FIXME: better solution?
+              this.#pathMap.set(value, path);
+              readCtrl.enqueue(value);
+            };
+          },
+        })
+      },
+      write: (chunk) => {
+        parser.write(chunk);
+      },
+    });
+    this.#readable = readable!; // sus
+  }
+
   #filterStream(expr: string) {
     return new TransformStream({
       transform: (value, controller) => {
         const path = this.#pathMap.get(value)!
         if (match(expr, path)) {
           controller.enqueue(value as any);
         } 
-        // Closing the stream early when the selected path can no longer yield values.
         else if (expr.startsWith(path)) {
+          // Closing the stream early when the selected path can no longer yield values.
           controller.terminate()
           // this.#streams.delete(expr) // no longer need to track the stream
         }
       }
     })
   }
 
-  // FIXME: Just acquiring this property will lock the internal stream. Different from regular transform stream.
   get readable(): ReadableStream<T> {
-    return this.#readable.pipeThrough(this.#filterStream(this.#jsonPath))
-  }
-
-  #a?: ReadableStream<T>;
-  get #readable() {
-    return this.#a ?? super.readable
+    return this.#readable
   }
 
   #clone() {
     const [a, b] = this.#readable.tee()
-    this.#a = a;
+    this.#readable = a;
     return b;
   }
 

test/index.test.ts

@@ -16,6 +16,7 @@ import * as json from '../index.ts';
 
 test('transform streams', () => {
   assertExists(json.JSONParseStream)
+  assertExists(json.JSONParseWritable)
   assertExists(json.JSONStringifyStream)
 })
 

test/json-parse-stream.test.ts

@@ -13,7 +13,7 @@ import {
 const { test } = Deno;
 
 import { jsonStringifyGenerator } from '../json-stringify.ts'
-import { JSONParseStream } from '../json-parse-stream.ts'
+import { JSONParseStream, JSONParseWritable } from '../json-parse-stream.ts'
 
 async function consume(stream: ReadableStream) {
   const reader = stream.getReader();
@@ -75,67 +75,65 @@ const aJoin = async (iter: AsyncIterable<string>, separator = '') => {
 }
 
 test('promise value', async () => {
-  const parseStream = new JSONParseStream()
+  const parseStream = new JSONParseWritable()
   const actual = {
     type: parseStream.promise('$.type'),
     data: parseStream.iterable('$.data.*')
   }
   const expected = JSON.stringify({ type: 'foo', data: [{ a: 1 }, { b: 2 }, { c: 3 }] })
   const done = new Response(expected).body!
-    .pipeThrough(parseStream)
-    .pipeTo(new WritableStream())
+    .pipeTo(parseStream)
 
   const actualString = await aJoin(jsonStringifyGenerator(actual))
   assertEquals(actualString, expected)
   await done;
 })
 
 test('promise value II', async () => {
-  const parseStream = new JSONParseStream()
+  const parseStream = new JSONParseWritable()
   const actual = {
     type: parseStream.promise('$.type'),
     data: parseStream.stream('$.data.*')
   }
   const expected = JSON.stringify({ type: 'foo', data: [{ a: 1 }, { b: 2 }, { c: 3 }] })
   const done = new Response(expected).body!
-    .pipeThrough(parseStream)
-    .pipeTo(new WritableStream())
+    .pipeTo(parseStream)
 
   const actualString = await aJoin(jsonStringifyGenerator(actual))
   assertEquals(actualString, expected)
   await done;
 })
 
 test('promise value III', async () => {
-  const parseStream = new JSONParseStream('$..*')
+  const parseStream = new JSONParseWritable()
   const actual = {
     type: parseStream.promise('$.type'),
     data: parseStream.iterable('$.data.*')
   }
   const expected = JSON.stringify({ type: 'foo', data: [{ a: 1 }, { b: 2 }, { c: 3 }] })
-  new Response(expected).body!.pipeThrough(parseStream) // no pipeTo necessary
+  new Response(expected).body!.pipeTo(parseStream)
 
   const actualString = await aJoin(jsonStringifyGenerator(actual))
   assertEquals(actualString, expected)
 })
 
 test('promise value IV', async () => {
-  const parseStream = new JSONParseStream('$..*')
+  const parseStream = new JSONParseWritable()
   const actual = {
     type: parseStream.promise('$.type'),
     data: parseStream.stream('$.data.*')
   }
   const expected = JSON.stringify({ type: 'foo', data: [{ a: 1 }, { b: 2 }, { c: 3 }] })
-  new Response(expected).body!.pipeThrough(parseStream) // no pipeTo necessary
+  new Response(expected).body!.pipeTo(parseStream)
 
   const actualString = await aJoin(jsonStringifyGenerator(actual))
   assertEquals(actualString, expected)
 })
 
 test('read only until first value', async () => {
-  const parseStream = new JSONParseStream('$..*')
+  const parseStream = new JSONParseWritable()
   const type = parseStream.promise('$.type');
   const expected = JSON.stringify({ type: 'foo', data: [{ a: 1 }, { b: 2 }, { c: 3 }] })
-  new Response(expected).body!.pipeThrough(parseStream) // no pipeTo necessary
+  new Response(expected).body!.pipeTo(parseStream)
   assertEquals(await type, 'foo')
 })

test/json-streams.test.ts test/playground.test.tstest/json-streams.test.ts renamed to test/playground.test.ts

@@ -13,7 +13,7 @@ import {
 const { test } = Deno;
 
 import { JSONStringifyStream } from '../json-stringify-stream.ts'
-import { JSONParseStream } from '../json-parse-stream.ts'
+import { JSONParseStream, JSONParseWritable } from '../json-parse-stream.ts'
 import { jsonStringifyStream } from '../json-stringify.ts'
 
 const collect = async <T>(stream: ReadableStream<T>) => {
@@ -66,7 +66,7 @@ test('roundtrip', async () => {
 })
 
 test('Retrieving multiple values and collections', async () => {
-  const jsonStream = new JSONParseStream();
+  const jsonStream = new JSONParseWritable();
   const asyncData = {
     type: jsonStream.promise('$.type'),
     items: jsonStream.stream('$.items.*'),
@@ -82,8 +82,7 @@ test('Retrieving multiple values and collections', async () => {
     ]
   };
 
-  new Response(JSON.stringify(nested)).body!
-    .pipeThrough(jsonStream) 
+  new Response(JSON.stringify(nested)).body!.pipeTo(jsonStream) 
 
   assertEquals(await asyncData.type, 'foo')