Merge pull request #42 from sergeyzenchenko/master

 Add array stream decoding

Author: gfx

benchmark/sync-vs-async.ts

@@ -1,7 +1,7 @@
 #!ts-node
 /* eslint-disable no-console */
 
-import { encode, decode, decodeAsync } from "../src";
+import { encode, decode, decodeAsync, decodeArrayStream } from "../src";
 import { writeFileSync, unlinkSync, readFileSync, createReadStream } from "fs";
 import { deepStrictEqual } from "assert";
 
@@ -39,4 +39,14 @@ import { deepStrictEqual } from "assert";
     await decodeAsync(createReadStream(file));
   }
   console.timeEnd("creteReadStream |> decodeAsync");
+
+  // asyncArrayStream
+
+  console.time("creteReadStream |> decodeArrayStream");
+  for (let i = 0; i < 100; i++) {
+    for await (let result of decodeArrayStream(createReadStream(file))) {
+      // console.log(result);
+    }
+  }
+  console.timeEnd("creteReadStream |> decodeArrayStream");
 })();

src/Decoder.ts

@@ -74,6 +74,20 @@ export class Decoder {
     this.pos = 0;
   }
 
+  appendBuffer(buffer: Uint8Array | ArrayLike<number>) {
+    if (this.headByte === HEAD_BYTE_REQUIRED && !this.hasRemaining()) {
+      this.setBuffer(buffer);
+    } else {
+      // retried because data is insufficient
+      const remainingData = this.bytes.subarray(this.pos);
+      const newData = ensureUint8Array(buffer);
+      const concated = new Uint8Array(remainingData.length + newData.length);
+      concated.set(remainingData);
+      concated.set(newData, remainingData.length);
+      this.setBuffer(concated);
+    }
+  }
+
   hasRemaining(size = 1) {
     return this.view.byteLength - this.pos >= size;
   }
@@ -99,18 +113,7 @@ export class Decoder {
         throw this.createNoExtraBytesError(this.totalPos);
       }
 
-      if (this.headByte === HEAD_BYTE_REQUIRED && !this.hasRemaining()) {
-        this.setBuffer(buffer);
-      } else {
-        // retried because data is insufficient
-        const remainingData = this.bytes.subarray(this.pos);
-        const newData = ensureUint8Array(buffer);
-        const concated = new Uint8Array(remainingData.length + newData.length);
-        concated.set(remainingData);
-        concated.set(newData, remainingData.length);
-        this.setBuffer(concated);
-      }
-      //console.log("view", this.view, this.headByte);
+      this.appendBuffer(buffer);
 
       try {
         object = this.decodeSync();
@@ -137,6 +140,47 @@ export class Decoder {
     );
   }
 
+  async *decodeArrayStream(stream: AsyncIterable<ArrayLike<number> | Uint8Array>) {
+    let headerParsed = false;
+    let decoded = false;
+    let itemsLeft = 0;
+
+    for await (const buffer of stream) {
+      if (decoded) {
+        throw this.createNoExtraBytesError(this.totalPos);
+      }
+
+      this.appendBuffer(buffer);
+
+      if (!headerParsed) {
+        itemsLeft = this.readArraySize();
+        headerParsed = true;
+        this.complete();
+      }
+
+      try {
+        while (true) {
+          let result = this.decodeSync();
+
+          yield result;
+
+          itemsLeft--;
+
+          if (itemsLeft === 0) {
+            decoded = true;
+            break;
+          }
+        }
+      } catch (e) {
+        if (!(e instanceof DataViewIndexOutOfBoundsError)) {
+          throw e; // rethrow
+        }
+        // fallthrough
+      }
+      this.totalPos += this.pos;
+    }
+  }
+
   decodeSync(): unknown {
     DECODE: while (true) {
       const headByte = this.readHeadByte();
@@ -363,6 +407,24 @@ export class Decoder {
     this.headByte = HEAD_BYTE_REQUIRED;
   }
 
+  readArraySize(): number {
+    const headByte = this.readHeadByte();
+
+    switch (headByte) {
+      case 0xdc:
+        return this.readU16();
+      case 0xdd:
+        return this.readU32();
+      default: {
+        if (headByte < 0xa0) {
+          return headByte - 0x90;
+        } else {
+          throw new Error(`Unrecognized array type byte: ${prettyByte(headByte)}`);
+        }
+      }
+    }
+  }
+
   pushMapState(size: number) {
     if (size > this.maxMapLength) {
       throw new Error(`Max length exceeded: map length (${size}) > maxMapLengthLength (${this.maxMapLength})`);

src/decodeAsync.ts

@@ -1,4 +1,3 @@
-import { ExtensionCodecType } from "./ExtensionCodec";
 import { Decoder } from "./Decoder";
 import { defaultDecodeOptions, DecodeOptions } from "./decode";
 
@@ -19,3 +18,21 @@ export async function decodeAsync(
   );
   return decoder.decodeOneAsync(stream);
 }
+
+export async function* decodeArrayStream(
+  stream: AsyncIterable<Uint8Array | ArrayLike<number>>,
+  options: DecodeAsyncOptions = defaultDecodeOptions,
+) {
+  const decoder = new Decoder(
+    options.extensionCodec,
+    options.maxStrLength,
+    options.maxBinLength,
+    options.maxArrayLength,
+    options.maxMapLength,
+    options.maxExtLength,
+  );
+
+  for await (let item of decoder.decodeArrayStream(stream)) {
+    yield item;
+  }
+}

src/index.ts

@@ -2,7 +2,7 @@
 
 export { encode } from "./encode";
 export { decode } from "./decode";
-export { decodeAsync } from "./decodeAsync";
+export { decodeAsync, decodeArrayStream } from "./decodeAsync";
 
 // Utilitiies for Extension Types: