From aa4ee9dadf99da15b7838eaf3d1fd5d0ff9ac679 Mon Sep 17 00:00:00 2001 From: crowlkats Date: Mon, 7 Mar 2022 16:19:17 +0100 Subject: [PATCH 1/3] limitedtransformstream --- streams/buffer.ts | 41 +++++++++++++++++ streams/buffer_test.ts | 102 ++++++++++++++++++++++++++++++++++++++--- 2 files changed, 137 insertions(+), 6 deletions(-) diff --git a/streams/buffer.ts b/streams/buffer.ts index 1e5524ebd801..3340d8114d8b 100644 --- a/streams/buffer.ts +++ b/streams/buffer.ts @@ -166,3 +166,44 @@ export class Buffer { this.#reslice(m); } } + +export class LimitedBytesTransformStream + extends TransformStream { + #read = 0; + constructor(size: number, options: { error?: boolean } = {}) { + super({ + transform: (chunk, controller) => { + if ((this.#read + chunk.byteLength) > size) { + if (options.error) { + throw new RangeError(`Exceeded byte size limit of '${size}'`); + } else { + controller.terminate(); + } + } else { + this.#read += chunk.byteLength; + controller.enqueue(chunk); + } + }, + }); + } +} + +export class LimitedTransformStream extends TransformStream { + #read = 0; + constructor(size: number, options: { error?: boolean } = {}) { + super({ + transform: (chunk, controller) => { + if ((this.#read + 1) > size) { + if (options.error) { + throw new RangeError(`Exceeded chunk limit of '${size}'`); + } else { + controller.terminate(); + } + } else { + this.#read++; + controller.enqueue(chunk); + } + }, + }); + } +} diff --git a/streams/buffer_test.ts b/streams/buffer_test.ts index 9f380f6db0bb..367e2db5df8d 100644 --- a/streams/buffer_test.ts +++ b/streams/buffer_test.ts @@ -1,9 +1,13 @@ // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license. -import { assert, assertEquals } from "../testing/asserts.ts"; -import { Buffer } from "./buffer.ts"; +import { assert, assertEquals, assertRejects } from "../testing/asserts.ts"; +import { + Buffer, + LimitedBytesTransformStream, + LimitedTransformStream, +} from "./buffer.ts"; -Deno.test("Buffer Write & Read", async function () { +Deno.test("[streams] Buffer Write & Read", async function () { const buf = new Buffer(); const writer = buf.writable.getWriter(); const reader = buf.readable.getReader({ mode: "byob" }); @@ -13,7 +17,7 @@ Deno.test("Buffer Write & Read", async function () { assertEquals(read.value, data); }); -Deno.test("Buffer Read empty", async function () { +Deno.test("[streams] Buffer Read empty", async function () { const buf = new Buffer(); const reader = buf.readable.getReader({ mode: "byob" }); const read = await reader.read(new Uint8Array(5)); @@ -21,7 +25,7 @@ Deno.test("Buffer Read empty", async function () { assertEquals(read.value!.byteLength, 0); }); -Deno.test("Buffer Write & get bytes", async function () { +Deno.test("[streams] Buffer Write & get bytes", async function () { const buf = new Buffer(); const writer = buf.writable.getWriter(); const data = new Uint8Array([4, 21, 45, 19]); @@ -29,10 +33,96 @@ Deno.test("Buffer Write & get bytes", async function () { assertEquals(buf.bytes(), data); }); -Deno.test("Buffer truncate", async function () { +Deno.test("[streams] Buffer truncate", async function () { const buf = new Buffer(); const writer = buf.writable.getWriter(); await writer.write(new Uint8Array([4, 21, 45, 19])); buf.truncate(3); assertEquals(buf.bytes(), new Uint8Array([4, 21, 45])); }); + +Deno.test("[streams] LimitedBytesTransformStream", async function () { + const r = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + controller.enqueue(new Uint8Array([4, 5, 6])); + controller.enqueue(new Uint8Array([7, 8, 9])); + controller.enqueue(new Uint8Array([10, 11, 12])); + controller.enqueue(new Uint8Array([13, 14, 15])); + controller.enqueue(new Uint8Array([16, 17, 18])); + controller.close(); + }, + }); + + const chunks = []; + for await (const chunk of r.pipeThrough(new LimitedBytesTransformStream(7))) { + chunks.push(chunk); + } + assertEquals(chunks.length, 2); +}); + +Deno.test("[streams] LimitedBytesTransformStream error", async function () { + const r = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + controller.enqueue(new Uint8Array([4, 5, 6])); + controller.enqueue(new Uint8Array([7, 8, 9])); + controller.enqueue(new Uint8Array([10, 11, 12])); + controller.enqueue(new Uint8Array([13, 14, 15])); + controller.enqueue(new Uint8Array([16, 17, 18])); + controller.close(); + }, + }); + + await assertRejects(async () => { + for await ( + const chunk of r.pipeThrough( + new LimitedBytesTransformStream(7, { error: true }), + ) + ) { + } + }, RangeError); +}); + +Deno.test("[streams] LimitedTransformStream", async function () { + const r = new ReadableStream({ + start(controller) { + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.close(); + }, + }); + + const chunks = []; + for await (const chunk of r.pipeThrough(new LimitedTransformStream(3))) { + chunks.push(chunk); + } + assertEquals(chunks.length, 3); +}); + +Deno.test("[streams] LimitedTransformStream error", async function () { + const r = new ReadableStream({ + start(controller) { + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.enqueue("foo"); + controller.close(); + }, + }); + + await assertRejects(async () => { + for await ( + const chunk of r.pipeThrough( + new LimitedTransformStream(3, { error: true }), + ) + ) { + } + }, RangeError); +}); From b6d4ca3aa664a37bfa3beeaf5a445ff0c850a16d Mon Sep 17 00:00:00 2001 From: crowlkats Date: Mon, 7 Mar 2022 16:24:47 +0100 Subject: [PATCH 2/3] lint --- streams/buffer_test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/streams/buffer_test.ts b/streams/buffer_test.ts index 367e2db5df8d..8e4295425606 100644 --- a/streams/buffer_test.ts +++ b/streams/buffer_test.ts @@ -76,10 +76,11 @@ Deno.test("[streams] LimitedBytesTransformStream error", async function () { await assertRejects(async () => { for await ( - const chunk of r.pipeThrough( + const _chunk of r.pipeThrough( new LimitedBytesTransformStream(7, { error: true }), ) ) { + // needed to read } }, RangeError); }); @@ -119,10 +120,11 @@ Deno.test("[streams] LimitedTransformStream error", async function () { await assertRejects(async () => { for await ( - const chunk of r.pipeThrough( + const _chunk of r.pipeThrough( new LimitedTransformStream(3, { error: true }), ) ) { + // needed to read } }, RangeError); }); From 6f524038c0813b93f70a09066029659a5a816e69 Mon Sep 17 00:00:00 2001 From: crowlkats Date: Mon, 7 Mar 2022 17:22:17 +0100 Subject: [PATCH 3/3] add comments --- streams/buffer.ts | 25 +++++++++++++++++++++++++ streams/mod.ts | 1 + 2 files changed, 26 insertions(+) diff --git a/streams/buffer.ts b/streams/buffer.ts index 3340d8114d8b..9078563df23c 100644 --- a/streams/buffer.ts +++ b/streams/buffer.ts @@ -167,6 +167,20 @@ export class Buffer { } } +/** A TransformStream that will only read & enqueue `size` amount of bytes. + * This operation is chunk based and not BYOB based, + * and as such will read more than needed. + * + * if options.error is set, then instead of terminating the stream, + * an error will be thrown. + * + * ```ts + * import { LimitedBytesTransformStream } from "./buffer.ts"; + * const res = await fetch("https://example.com"); + * const parts = res.body! + * .pipeThrough(new LimitedBytesTransformStream(512 * 1024)); + * ``` + */ export class LimitedBytesTransformStream extends TransformStream { #read = 0; @@ -188,6 +202,17 @@ export class LimitedBytesTransformStream } } +/** A TransformStream that will only read & enqueue `size` amount of chunks. + * + * if options.error is set, then instead of terminating the stream, + * an error will be thrown. + * + * ```ts + * import { LimitedTransformStream } from "./buffer.ts"; + * const res = await fetch("https://example.com"); + * const parts = res.body!.pipeThrough(new LimitedTransformStream(50)); + * ``` + */ export class LimitedTransformStream extends TransformStream { #read = 0; constructor(size: number, options: { error?: boolean } = {}) { diff --git a/streams/mod.ts b/streams/mod.ts index 1475344c5bb5..daf3a23889f4 100644 --- a/streams/mod.ts +++ b/streams/mod.ts @@ -1,5 +1,6 @@ // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license. +export * from "./buffer.ts"; export * from "./conversion.ts"; export * from "./delimiter.ts"; export * from "./merge.ts";