Use chunk utility on streams.

Author: octet-stream

package.json

@@ -20,8 +20,12 @@
   "main": "./lib/index.js",
   "module": "./lib/index.js",
   "exports": {
-    "types": "./@type/index.d.ts",
-    "default": "./lib/index.js"
+    "./package.json": "./package.json",
+    ".": {
+      "types": "./@type/index.d.ts",
+      "import": "./lib/index.js",
+      "default": "./lib/index.js"
+    }
   },
   "types": "./@type/index.d.ts",
   "scripts": {

src/util/chunk.ts

@@ -1,5 +1,17 @@
 export const MAX_CHUNK_SIZE = 65536
 
+/**
+ * Chunks given `value` into evenly sized pieces (Same as `MAX_CHUNK_SIZE` bytes per each).
+ * Returns `Generator<Uint8Array>` allowing to iterate over these chunks.
+ *
+ * If value is less then `MAX_CHUNK_SIZE`, it will be returned as-is.
+ *
+ * If the last chunk is less than `MAX_CHUNK_SIZE`, then returned value will be the size that chunk.
+ *
+ * @param value A value to chunk into evenly sized pieces
+ *
+ * @api private
+ */
 export function* chunk(value: Uint8Array): Generator<Uint8Array, void> {
   if (value.byteLength <= MAX_CHUNK_SIZE) {
     yield value

src/util/getStreamIterator.test.ts

@@ -4,30 +4,44 @@ import {ReadableStream} from "web-streams-polyfill"
 import {stub} from "sinon"
 
 import {getStreamIterator} from "./getStreamIterator.js"
+import {isAsyncIterable} from "./isAsyncIterable.js"
 
-test(
-  "Returns readable stream as is, if it implements Symbol.asyncIterator",
+test("Returns async iterable for streams w/ Symbol.asyncIterator", t => {
+  const stream = new ReadableStream()
 
-  t => {
-    const stream = new ReadableStream()
+  t.true(isAsyncIterable(getStreamIterator(stream)))
+})
+
+test("Iterates over given stream", async t => {
+  const expected = "Some text"
 
-    t.is(getStreamIterator(stream), stream)
+  const stream = new ReadableStream({
+    pull(controller) {
+      controller.enqueue(new TextEncoder().encode(expected))
+      controller.close()
+    }
+  })
+
+  let actual = ""
+  const decoder = new TextDecoder()
+  for await (const chunk of getStreamIterator(stream)) {
+    actual += decoder.decode(chunk, {stream: true})
   }
-)
 
-test(
-  "Returns fallback when given stream does not implement Symbol.asyncIterator",
+  actual += decoder.decode()
+
+  t.is(actual, expected)
+})
 
-  t => {
-    const stream = new ReadableStream()
+test("Returns async iterable for streams w/o Symbol.asyncIterator", t => {
+  const stream = new ReadableStream()
 
-    stub(stream, Symbol.asyncIterator).get(() => undefined)
+  stub(stream, Symbol.asyncIterator).get(() => undefined)
 
-    t.false(getStreamIterator(stream) instanceof ReadableStream)
-  }
-)
+  t.false(getStreamIterator(stream) instanceof ReadableStream)
+})
 
-test("Reads from the stream using fallback", async t => {
+test("Iterates over the stream using fallback", async t => {
   const expected = "Some text"
 
   const stream = new ReadableStream({

src/util/getStreamIterator.ts

@@ -1,13 +1,6 @@
+import {isAsyncIterable} from "./isAsyncIterable.js"
 import {isFunction} from "./isFunction.js"
-
-/**
- * Checks if the value is async iterable
- */
-const isAsyncIterable = (
-  value: unknown
-): value is AsyncIterable<Uint8Array> => (
-  isFunction((value as AsyncIterable<Uint8Array>)[Symbol.asyncIterator])
-)
+import {chunk} from "./chunk.js"
 
 /**
  * Reads from given ReadableStream
@@ -30,6 +23,14 @@ async function* readStream(
   }
 }
 
+async function* chunkStream(
+  stream: AsyncIterable<Uint8Array>
+): AsyncGenerator<Uint8Array, void> {
+  for await (const value of stream) {
+    yield* chunk(value)
+  }
+}
+
 /**
  * Turns ReadableStream into async iterable when the `Symbol.asyncIterable` is not implemented on given stream.
  *
@@ -39,11 +40,11 @@ export const getStreamIterator = (
   source: ReadableStream<Uint8Array> | AsyncIterable<Uint8Array>
 ): AsyncIterable<Uint8Array> => {
   if (isAsyncIterable(source)) {
-    return source
+    return chunkStream(source)
   }
 
   if (isFunction(source.getReader)) {
-    return readStream(source)
+    return chunkStream(readStream(source))
   }
 
   // Throw an error otherwise (for example, in case if encountered Node.js Readable stream without Symbol.asyncIterator method)

src/util/isAsyncIterable.ts

@@ -0,0 +1,10 @@
+import {isFunction} from "./isFunction.js"
+
+/**
+ * Checks if the value is async iterable
+ */
+export const isAsyncIterable = (
+  value: object
+): value is AsyncIterable<Uint8Array> => (
+  isFunction((value as AsyncIterable<Uint8Array>)[Symbol.asyncIterator])
+)