All files / cbor / _map_decoded_stream.ts

66.67% Branches 2/3
95.00% Lines 19/20
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
x40
 
 
 
 
 
 
x40
x40
x40
x40
x61
x61
x152
x152
x172
x172
x152
x61
x61
 
 
x62
x62
x61
x61
x40





























































I




// Copyright 2018-2025 the Deno authors. MIT license.

import type { ReleaseLock } from "./_common.ts";
import type { CborMapStreamOutput } from "./types.ts";

/**
 * A {@link ReadableStream} that wraps the decoded CBOR "Map".
 * [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 {
 *   CborMapDecodedStream,
 *   CborMapEncoderStream,
 *   CborSequenceDecoderStream,
 * } from "@std/cbor";
 *
 * const rawMessage: Record<string, number> = {
 *   a: 0,
 *   b: 1,
 *   c: 2,
 *   d: 3,
 * };
 *
 * for await (
 *   const value of ReadableStream.from(Object.entries(rawMessage))
 *     .pipeThrough(new CborMapEncoderStream)
 *     .pipeThrough(new CborSequenceDecoderStream())
 * ) {
 *   assert(value instanceof CborMapDecodedStream);
 *   for await (const [k, v] of value) {
 *     assertEquals(rawMessage[k], v);
 *   }
 * }
 * ```
 */
export class CborMapDecodedStream extends ReadableStream<CborMapStreamOutput> {
  /**
   * Constructs a new instance.
   *
   * @param gen A {@link AsyncGenerator<CborMapStreamOutput>}.
   * @param releaseLock A Function that's called when the stream is finished.
   */
  constructor(
    gen: AsyncGenerator<CborMapStreamOutput>,
    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();
      },
    });
  }
}