Add decodeStream (#46)

* Add decodeStream

* Fix code review comments

* Fix codereview

Author: sergeyzenchenko

src/Decoder.ts

@@ -140,6 +140,25 @@ export class Decoder {
     );
   }
 
+  async *decodeStream(stream: AsyncIterable<ArrayLike<number> | Uint8Array>) {
+    for await (const buffer of stream) {
+      this.appendBuffer(buffer);
+
+      try {
+        while (true) {
+          const result = this.decodeSync();
+
+          yield result;
+        }
+      } catch (e) {
+        if (!(e instanceof DataViewIndexOutOfBoundsError)) {
+          throw e; // rethrow
+        }
+        // fallthrough
+      }
+    }
+  }
+
   async *decodeArrayStream(stream: AsyncIterable<ArrayLike<number> | Uint8Array>) {
     let headerParsed = false;
     let decoded = false;
@@ -160,7 +179,7 @@ export class Decoder {
 
       try {
         while (true) {
-          let result = this.decodeSync();
+          const result = this.decodeSync();
 
           yield result;
 

src/decodeAsync.ts

@@ -22,7 +22,7 @@ export async function decodeAsync(
   return decoder.decodeOneAsync(stream);
 }
 
-export async function* decodeArrayStream(
+export function decodeArrayStream(
   streamLike: ReadableStreamLike<Uint8Array | ArrayLike<number>>,
   options: DecodeAsyncOptions = defaultDecodeOptions,
 ) {
@@ -37,7 +37,23 @@ export async function* decodeArrayStream(
     options.maxExtLength,
   );
 
-  for await (let item of decoder.decodeArrayStream(stream)) {
-    yield item;
-  }
+  return decoder.decodeArrayStream(stream);
+}
+
+export function decodeStream(
+  streamLike: ReadableStreamLike<Uint8Array | ArrayLike<number>>,
+  options: DecodeAsyncOptions = defaultDecodeOptions,
+) {
+  const stream = ensureAsyncIterabe(streamLike);
+
+  const decoder = new Decoder(
+    options.extensionCodec,
+    options.maxStrLength,
+    options.maxBinLength,
+    options.maxArrayLength,
+    options.maxMapLength,
+    options.maxExtLength,
+  );
+
+  return decoder.decodeStream(stream);
 }

test/decodeStream.test.ts

@@ -0,0 +1,30 @@
+import assert from "assert";
+import { encode } from "../src";
+import { decodeStream } from "../src/decodeAsync";
+
+describe("decodeStream", () => {
+  it("decodes stream", async () => {
+    const items = [
+      "foo",
+      10,
+      {
+        name: "bar",
+      },
+      [1, 2, 3],
+    ];
+
+    const createStream = async function*() {
+      for (const item of items) {
+        yield encode(item);
+      }
+    };
+
+    const result = [];
+
+    for await (const item of decodeStream(createStream())) {
+      result.push(item);
+    }
+
+    assert.deepStrictEqual(result, items);
+  });
+});