feat(cbor/unstable): introduce @std/cbor

_tools/check_circular_package_dependencies.ts

@@ -40,6 +40,7 @@ type Mod =
   | "async"
   | "bytes"
   | "cache"
+  | "cbor"
   | "cli"
   | "collections"
   | "crypto"
@@ -83,6 +84,7 @@ const ENTRYPOINTS: Record<Mod, string[]> = {
   async: ["mod.ts"],
   bytes: ["mod.ts"],
   cache: ["mod.ts"],
+  cbor: ["mod.ts"],
   cli: ["mod.ts"],
   collections: ["mod.ts"],
   crypto: ["mod.ts"],

_tools/check_docs.ts

@@ -35,6 +35,7 @@ const ENTRY_POINTS = [
   "../async/mod.ts",
   "../bytes/mod.ts",
   "../cache/mod.ts",
+  "../cbor/mod.ts",
   "../cli/mod.ts",
   "../cli/unstable_spinner.ts",
   "../crypto/mod.ts",

browser-compat.tsconfig.json

@@ -13,6 +13,7 @@
     "./async",
     "./bytes",
     "./cache",
+    "./cbor",
     "./cli",
     "./collections",
     "./crypto",

cbor/_array_decoded_stream.ts

@@ -0,0 +1,62 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+import type { ReleaseLock } from "./_common.ts";
+import type { CborStreamOutput } from "./types.ts";
+
+/**
+ * A {@link ReadableStream} that wraps the decoded CBOR "Array".
+ * [RFC 8949 - Concise Binary Object Representation (CBOR)](https://datatracker.ietf.org/doc/html/rfc8949)
+ *
+ * Instances of this class is created from {@link CborSequenceDecoderStream}.
+ * This class is not designed for you to create instances of it yourself. It is
+ * merely a way for you to validate the type being returned.
+ *
+ * @example Usage
+ * ```ts
+ * import { assert, assertEquals } from "@std/assert";
+ * import {
+ *   CborArrayDecodedStream,
+ *   CborArrayEncoderStream,
+ *   CborSequenceDecoderStream,
+ * } from "@std/cbor";
+ *
+ * const rawMessage = ["a".repeat(100), "b".repeat(100), "c".repeat(100)];
+ *
+ * for await (
+ *   const value of ReadableStream.from(rawMessage)
+ *     .pipeThrough(new CborArrayEncoderStream())
+ *     .pipeThrough(new CborSequenceDecoderStream())
+ * ) {
+ *   assert(value instanceof CborArrayDecodedStream);
+ *   let i = 0;
+ *   for await (const text of value) {
+ *     assert(typeof text === "string");
+ *     assertEquals(text, rawMessage[i++]);
+ *   }
+ * }
+ * ```
+ */
+export class CborArrayDecodedStream extends ReadableStream<CborStreamOutput> {
+  /**
+   * Constructs a new instance.
+   *
+   * @param gen A {@link AsyncGenerator<CborStreamOutput>}.
+   * @param releaseLock A Function that's called when the stream is finished.
+   */
+  constructor(gen: AsyncGenerator<CborStreamOutput>, releaseLock: ReleaseLock) {
+    super({
+      async pull(controller) {
+        const { done, value } = await gen.next();
+        if (done) {
+          releaseLock();
+          controller.close();
+        } else controller.enqueue(value);
+      },
+      async cancel() {
+        // deno-lint-ignore no-empty
+        for await (const _ of gen) {}
+        releaseLock();
+      },
+    });
+  }
+}

cbor/_array_decoded_stream_test.ts

@@ -0,0 +1,48 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+import { assert, assertEquals } from "@std/assert";
+import { random } from "./_common_test.ts";
+import { CborArrayDecodedStream } from "./_array_decoded_stream.ts";
+import { encodeCbor } from "./encode_cbor.ts";
+import { CborSequenceDecoderStream } from "./sequence_decoder_stream.ts";
+
+Deno.test("CborArrayDecodedStream() being consumed", async () => {
+  const size = random(0, 24);
+
+  const reader = ReadableStream.from([encodeCbor(new Array(size).fill(0))])
+    .pipeThrough(new CborSequenceDecoderStream()).getReader();
+
+  const { done, value } = await reader.read();
+  assert(done === false);
+  assert(value instanceof CborArrayDecodedStream);
+  assertEquals(await Array.fromAsync(value), new Array(size).fill(0));
+
+  assert((await reader.read()).done === true);
+  reader.releaseLock();
+});
+
+Deno.test("CborArrayDecodedStream() being cancelled", async () => {
+  const size = random(0, 24);
+  const reader = ReadableStream.from([
+    encodeCbor(new Array(size).fill(0)),
+    encodeCbor(0),
+  ])
+    .pipeThrough(new CborSequenceDecoderStream()).getReader();
+
+  {
+    const { done, value } = await reader.read();
+    assert(done === false);
+    assert(value instanceof CborArrayDecodedStream);
+    await value.cancel();
+  }
+
+  {
+    const { done, value } = await reader.read();
+    assert(done === false);
+    assert(typeof value === "number");
+    assertEquals(value, 0);
+  }
+
+  assert((await reader.read()).done === true);
+  reader.releaseLock();
+});

cbor/_byte_decoded_stream.ts

@@ -0,0 +1,60 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+import type { ReleaseLock } from "./_common.ts";
+
+/**
+ * A {@link ReadableStream} that wraps the decoded CBOR "Byte String".
+ * [RFC 8949 - Concise Binary Object Representation (CBOR)](https://datatracker.ietf.org/doc/html/rfc8949)
+ *
+ * Instances of this class is created from {@link CborSequenceDecoderStream}.
+ * This class is not designed for you to create instances of it yourself. It is
+ * merely a way for you to validate the type being returned.
+ *
+ * @example Usage
+ * ```ts
+ * import { assert, assertEquals } from "@std/assert";
+ * import { concat } from "@std/bytes";
+ * import {
+ *   CborByteDecodedStream,
+ *   CborByteEncoderStream,
+ *   CborSequenceDecoderStream,
+ * } from "@std/cbor";
+ *
+ * const rawMessage = new Uint8Array(100);
+ *
+ * for await (
+ *   const value of ReadableStream.from([rawMessage])
+ *     .pipeThrough(new CborByteEncoderStream())
+ *     .pipeThrough(new CborSequenceDecoderStream())
+ * ) {
+ *   assert(value instanceof Uint8Array || value instanceof CborByteDecodedStream);
+ *   if (value instanceof CborByteDecodedStream) {
+ *     assertEquals(concat(await Array.fromAsync(value)), new Uint8Array(100));
+ *   } else assertEquals(value, new Uint8Array(100));
+ * }
+ * ```
+ */
+export class CborByteDecodedStream extends ReadableStream<Uint8Array> {
+  /**
+   * Constructs a new instance.
+   *
+   * @param gen A {@link AsyncGenerator<Uint8Array>}.
+   * @param releaseLock A Function that's called when the stream is finished.
+   */
+  constructor(gen: AsyncGenerator<Uint8Array>, releaseLock: ReleaseLock) {
+    super({
+      async pull(controller) {
+        const { done, value } = await gen.next();
+        if (done) {
+          releaseLock();
+          controller.close();
+        } else controller.enqueue(value);
+      },
+      async cancel() {
+        // deno-lint-ignore no-empty
+        for await (const _ of gen) {}
+        releaseLock();
+      },
+    });
+  }
+}

cbor/_byte_decoded_stream_test.ts

@@ -0,0 +1,80 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+import { assert, assertEquals } from "@std/assert";
+import { concat } from "@std/bytes";
+import { random } from "./_common_test.ts";
+import { CborByteDecodedStream } from "./_byte_decoded_stream.ts";
+import { CborByteEncoderStream } from "./byte_encoder_stream.ts";
+import { encodeCbor } from "./encode_cbor.ts";
+import { CborSequenceDecoderStream } from "./sequence_decoder_stream.ts";
+import { CborSequenceEncoderStream } from "./sequence_encoder_stream.ts";
+
+Deno.test("CborByteDecodedStream() consuming indefinite length byte string", async () => {
+  const size = random(0, 24);
+
+  const reader = CborByteEncoderStream.from([
+    new Uint8Array(size),
+    new Uint8Array(size * 2),
+    new Uint8Array(size * 3),
+  ]).readable.pipeThrough(new CborSequenceDecoderStream()).getReader();
+
+  const { done, value } = await reader.read();
+  assert(done === false);
+  assert(value instanceof CborByteDecodedStream);
+  assertEquals(await Array.fromAsync(value), [
+    new Uint8Array(size),
+    new Uint8Array(size * 2),
+    new Uint8Array(size * 3),
+  ]);
+
+  assert((await reader.read()).done === true);
+  reader.releaseLock();
+});
+
+Deno.test("CborByteDecodedStream() consuming large definite length byte string", async () => {
+  // Uint8Array needs to be 2 ** 32 bytes+ to be decoded via a CborByteDecodedStream.
+  const size = random(2 ** 32, 2 ** 33);
+
+  const reader = ReadableStream.from([encodeCbor(new Uint8Array(size))])
+    .pipeThrough(new CborSequenceDecoderStream()).getReader();
+
+  const { done, value } = await reader.read();
+  assert(done === false);
+  assert(value instanceof CborByteDecodedStream);
+  assertEquals(concat(await Array.fromAsync(value)).length, size);
+
+  assert((await reader.read()).done === true);
+  reader.releaseLock();
+});
+
+Deno.test("CborByteDecodedStream() being cancelled", async () => {
+  const size = random(0, 24);
+
+  const reader = ReadableStream.from([
+    CborByteEncoderStream.from([
+      new Uint8Array(size),
+      new Uint8Array(size * 2),
+      new Uint8Array(size * 3),
+    ]),
+    0,
+  ])
+    .pipeThrough(new CborSequenceEncoderStream())
+    .pipeThrough(new CborSequenceDecoderStream()).getReader();
+
+  {
+    const { done, value } = await reader.read();
+    assert(done === false);
+    assert(value instanceof CborByteDecodedStream);
+    await value.cancel();
+  }
+
+  {
+    const { done, value } = await reader.read();
+    assert(done === false);
+    assert(typeof value === "number");
+    assertEquals(value, 0);
+  }
+
+  assert((await reader.read()).done === true);
+  reader.releaseLock();
+});

cbor/_common.ts

@@ -0,0 +1,92 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+export type ReleaseLock = (value?: unknown) => void;
+
+export function numberToArray(bytes: number, x: number | bigint): Uint8Array {
+  const view = new DataView(new ArrayBuffer(8));
+  if (typeof x === "bigint" || x % 1 === 0) view.setBigUint64(0, BigInt(x));
+  else view.setFloat64(0, x);
+  return new Uint8Array(view.buffer.slice(-bytes));
+}
+
+export function arrayToNumber(
+  buffer: ArrayBufferLike & { BYTES_PER_ELEMENT?: never },
+  isInteger: true,
+): number | bigint;
+export function arrayToNumber(
+  buffer: ArrayBufferLike & { BYTES_PER_ELEMENT?: never },
+  isInteger: false,
+): number;
+export function arrayToNumber(
+  buffer: ArrayBufferLike & { BYTES_PER_ELEMENT?: never },
+  isInteger: boolean,
+): number | bigint {
+  const view = new DataView(buffer);
+  if (isInteger) {
+    switch (buffer.byteLength) {
+      case 1:
+        return view.getUint8(0);
+      case 2:
+        return view.getUint16(0);
+      case 4:
+        return view.getUint32(0);
+      default:
+        return view.getBigUint64(0);
+    }
+  }
+  switch (buffer.byteLength) {
+    case 2:
+      return view.getFloat16(0);
+    case 4:
+      return view.getFloat32(0);
+    default:
+      return view.getFloat64(0);
+  }
+}
+
+// To be removed if https://github.com/denoland/std/pull/6046 merges.
+export function toByteStream(
+  readable: ReadableStream<Uint8Array>,
+): ReadableStream<Uint8Array> {
+  try {
+    const reader = readable.getReader({ mode: "byob" });
+    reader.releaseLock();
+    return readable;
+  } catch {
+    const reader = readable.getReader();
+    return new ReadableStream({
+      type: "bytes",
+      async pull(controller) {
+        const value = await async function () {
+          while (true) {
+            const { done, value } = await reader.read();
+            if (done) return undefined;
+            if (value.length) return value;
+          }
+        }();
+
+        if (value == undefined) {
+          controller.close();
+          return controller.byobRequest?.respond(0);
+        }
+
+        if (controller.byobRequest?.view) {
+          const buffer = new Uint8Array(controller.byobRequest.view.buffer);
+          const offset = controller.byobRequest.view.byteOffset;
+          const size = buffer.length - offset;
+          if (value.length > size) {
+            buffer.set(value.slice(0, size), offset);
+            controller.byobRequest.respond(size);
+            controller.enqueue(value.slice(size));
+          } else {
+            buffer.set(value, offset);
+            controller.byobRequest.respond(value.length);
+          }
+        } else controller.enqueue(value);
+      },
+      async cancel(reason) {
+        await reader.cancel(reason);
+      },
+    });
+  }
+}