feat: add function parseStream (#16)

* wip: parsing stream

* wip: first example with generator

* chore: allow to define CHUNK_SIZE in tests

* chore: fix parseStream chunk size

* wip: allow to join chunks

* wip: try to slice the data

* fix: throw error if no closing tag

* feat: add parseStream

* chore: test with node 18

* fix: only test parseStream with node 18 and greater

Author: lpatiny

.github/workflows/nodejs.yml

@@ -10,5 +10,5 @@ jobs:
   nodejs:
     uses: zakodium/workflows/.github/workflows/nodejs.yml@nodejs-v1
     with:
-      node-version-matrix: '[14, 16]'
+      node-version-matrix: '[14, 16, 18]'
       lint-check-types: true

.gitignore

@@ -119,4 +119,6 @@ dist
 
 lib
 lib-esm
-big.xml
+big.xml
+
+script/medline.xml

package.json

@@ -39,18 +39,18 @@
   "homepage": "https://github.com/cheminfo/arraybuffer-xml-parser#readme",
   "devDependencies": {
     "@types/he": "^1.1.2",
-    "@types/jest": "^27.5.0",
+    "@types/jest": "^27.5.1",
     "cheminfo-build": "^1.1.11",
-    "eslint": "^8.15.0",
+    "eslint": "^8.16.0",
     "eslint-config-cheminfo-typescript": "^10.4.0",
     "he": "^1.2.0",
     "iobuffer": "^5.1.0",
     "jest": "^28.1.0",
     "pako": "^2.0.4",
     "prettier": "^2.6.2",
     "rimraf": "^3.0.2",
-    "ts-jest": "^28.0.2",
-    "typescript": "^4.6.4",
+    "ts-jest": "^28.0.3",
+    "typescript": "^4.7.2",
     "uint8-base64": "^0.1.1"
   },
   "dependencies": {

script/medline.mjs

@@ -0,0 +1,19 @@
+import { parseStream } from '../lib/index.js';
+import { open } from 'fs/promises';
+
+/*
+In order to test this script you should first build the package: `npm run prepack`
+And you also need a (big) file from medline called 'medline.xml'
+*/
+
+async function doAll() {
+  const file = await open(new URL('medline.xml', import.meta.url), 'r');
+  const stream = file.readableWebStream();
+  let i = 0;
+  for await (const entry of parseStream(stream, 'PubmedArticle')) {
+    console.log(entry);
+    console.log(i++);
+  }
+}
+
+doAll();

src/__tests__/parseStream.test.ts

@@ -0,0 +1,63 @@
+import { open } from 'fs/promises';
+import { join } from 'path';
+
+import { parseStream } from '../parseStream';
+
+describe('parseStream', () => {
+  it('simple case', async () => {
+    // eslint-disable-next-line jest/no-if
+    if (Number(process.versions.node.split('.')[0]) >= 18) {
+      const file = await open(join(__dirname, 'assets/sample.xml'), 'r');
+      const CHUNK_SIZE = 10;
+      const transformStream = new TransformStream({
+        start: function start() {}, // required.
+        transform: async function transform(chunk, controller) {
+          if (chunk === null) controller.terminate();
+          chunk = new Uint8Array(await chunk);
+          for (let i = 0; i < chunk.length; i += CHUNK_SIZE) {
+            controller.enqueue(chunk.slice(i, i + CHUNK_SIZE));
+          }
+        },
+      });
+
+      const results = [];
+      //@ts-expect-error feature is too new
+      const readableStream = file.readableWebStream();
+      for await (let entry of parseStream(
+        readableStream.pipeThrough(transformStream),
+        'address',
+      )) {
+        results.push(entry);
+        //console.log(entry);
+      }
+      expect(results).toMatchInlineSnapshot(`
+      Array [
+        Object {
+          "buildingNo": 1,
+          "city": "New York",
+          "flatNo": 1,
+          "street": "Park Ave",
+        },
+        Object {
+          "buildingNo": 33,
+          "city": "Boston",
+          "flatNo": 24,
+          "street": "Centre St",
+        },
+        Object {
+          "buildingNo": 1,
+          "city": "Moscow",
+          "flatNo": 2,
+          "street": "Kahovka",
+        },
+        Object {
+          "buildingNo": 3,
+          "city": "Tula",
+          "flatNo": 78,
+          "street": "Lenina",
+        },
+      ]
+    `);
+    }
+  });
+});

src/index.ts

@@ -1 +1,2 @@
 export * from './parse';
+export * from './parseStream';

src/parseStream.ts

@@ -0,0 +1,25 @@
+import {
+  defaultOptions,
+  StreamParseOptions,
+} from './traversable/defaultOptions';
+import { getTraversableGenerator } from './traversable/getTraversableGenerator';
+import { traversableToJSON } from './traversableToJSON';
+
+/**
+ * Parse a web stream representing an XML and emit objects
+ */
+export async function* parseStream(
+  readableStream: ReadableStream,
+  lookupTagName: string,
+  options: StreamParseOptions = {},
+) {
+  options = { ...defaultOptions, ...options };
+
+  for await (const traversableEntry of getTraversableGenerator(
+    readableStream,
+    lookupTagName,
+    options,
+  )) {
+    yield traversableToJSON(traversableEntry, options);
+  }
+}

src/traversable/closingIndexForOpeningTag.ts

@@ -1,5 +1,11 @@
 import { decoder } from './utils/utf8Decoder';
 
+/**
+ * Search for the corresponding closing tag '>'
+ * @param data
+ * @param i
+ * @returns
+ */
 export function closingIndexForOpeningTag(
   data: Uint8Array,
   i: number,
@@ -25,8 +31,5 @@ export function closingIndexForOpeningTag(
     }
     endIndex++;
   }
-  return {
-    data: decoder.decode(data.subarray(i, i + endIndex)),
-    index: 0,
-  };
+  throw new Error('Could not find closing tag');
 }

src/traversable/defaultOptions.ts

@@ -7,6 +7,20 @@ export const decoder = {
     return utf8Decoder.decode(array);
   },
 };
+
+export interface StreamParseOptions extends ParseOptions {
+  /**
+   * What is the maximal size (in bytes) of an entry
+   * @default 1e7
+   */
+  maxEntrySize?: number;
+  /**
+   * What is the maximal size for the buffer
+   * @default 2e8
+   */
+  maxBufferSize?: number;
+}
+
 export interface ParseOptions {
   /**
    * should we remove ascii < 32
@@ -92,6 +106,7 @@ export interface ParseOptions {
    */
   stopNodes?: string[];
 }
+
 export const defaultOptions: ParseOptions = {
   trimValues: true,
   attributeNamePrefix: '$',