{"version":3,"file":"BaseStream.cjs","sourceRoot":"","sources":["../src/BaseStream.ts"],"names":[],"mappings":";;;;;;;;;;;;;;;AAAA,mDAAmD;AAEnD,yDAAmD;AAInD,0CAQoB;AAEpB,MAAM,gBAAgB,GAAG,GAGrB,EAAE;IACJ,MAAM,WAAW,GAAsB,EAAE,CAAC;IAC1C,MAAM,YAAY,GAAuB,EAAE,CAAC;IAC5C,IAAI,IAAI,GAAG,KAAK,CAAC;IAEjB,OAAO;QACL;;;;;;WAMG;QACH,GAAG,EAAE,CAAC,KAAa,EAAQ,EAAE;YAC3B,IAAI,IAAI,EAAE,CAAC;gBACT,OAAO;YACT,CAAC;YACD,IAAI,GAAG,IAAI,CAAC;YAEZ,KAAK,MAAM,EAAE,OAAO,EAAE,MAAM,EAAE,IAAI,YAAY,EAAE,CAAC;gBAC/C,KAAK,CAAC,CAAC,CAAC,MAAM,CAAC,KAAK,CAAC,CAAC,CAAC,CAAC,OAAO,CAAC,IAAA,yBAAc,GAAW,CAAC,CAAC;YAC7D,CAAC;YACD,YAAY,CAAC,MAAM,GAAG,CAAC,CAAC;QAC1B,CAAC;QAED,eAAe;YACb,OAAO,YAAY,CAAC,MAAM,GAAG,CAAC,CAAC;QACjC,CAAC;QAED;;;;;WAKG;QACH,GAAG,CAAC,KAAoB;YACtB,IAAI,IAAI,EAAE,CAAC;gBACT,OAAO;YACT,CAAC;YAED,IAAI,YAAY,CAAC,MAAM,GAAG,CAAC,EAAE,CAAC;gBAC5B,MAAM,EAAE,OAAO,EAAE,GAAG,YAAY,CAAC,KAAK,EAAsB,CAAC;gBAC7D,OAAO,CAAC,KAAK,CAAC,CAAC;gBACf,OAAO;YACT,CAAC;YACD,WAAW,CAAC,IAAI,CAAC,KAAK,CAAC,CAAC;QAC1B,CAAC;QAED,KAAK,CAAC,GAAG;YACP,IAAI,WAAW,CAAC,MAAM,GAAG,CAAC,EAAE,CAAC;gBAC3B,MAAM,KAAK,GAAG,WAAW,CAAC,KAAK,EAAW,CAAC;gBAC3C,OAAO,KAAK,YAAY,KAAK;oBAC3B,CAAC,CAAC,OAAO,CAAC,MAAM,CAAC,KAAK,CAAC;oBACvB,CAAC,CAAC,OAAO,CAAC,OAAO,CAAC,KAAK,CAAC,CAAC;YAC7B,CAAC;YAED,IAAI,IAAI,EAAE,CAAC;gBACT,OAAO,IAAA,yBAAc,GAAW,CAAC;YACnC,CAAC;YAED,MAAM,EAAE,OAAO,EAAE,OAAO,EAAE,MAAM,EAAE,GAAG,IAAA,4BAAc,GAAS,CAAC;YAC7D,YAAY,CAAC,IAAI,CAAC;gBAChB,OAAO,EAAE,OAAmC;gBAC5C,MAAM;aACP,CAAC,CAAC;YACH,OAAO,OAAO,CAAC;QACjB,CAAC;KACF,CAAC;AACJ,CAAC,CAAC;AACF,MAAM,CAAC,gBAAgB,CAAC,CAAC;AAyBzB;;;;;;;;;GASG;AACH,MAAa,UAAU;IAerB;;;;;;;;OAQG;IACH,YAAY,EAAE,IAAI,EAAE,KAAK,EAAE,aAAa,EAAwB;;QAvBhE;;;WAGG;QACM,6BAAU,gBAAgB,EAAmC,EAAC;QAE9D,mCAAc;QAEd,4CAAiD;QAE1D,oCAA2B;QAE3B,4CAAkC,KAAK,EAAC;QAkC/B,mCAAgB,KAAK,EAAE,KAAc,EAAiB,EAAE;YAC/D,6DAA6D;YAC7D,MAAM,IAAI,CAAC;YAEX,MAAM,WAAW,GAAG,IAAA,oBAAS,EAAC,KAAK,CAAC,CAAC;YACrC,IAAI,WAAW,YAAY,KAAK,EAAE,CAAC;gBACjC,MAAM,uBAAA,IAAI,2DAAkB,MAAtB,IAAI,EAAmB,WAAW,CAAC,CAAC;gBAC1C,OAAO;YACT,CAAC;YAED,IAAI,WAAW,KAAK,2BAAgB,EAAE,CAAC;gBACrC,MAAM,uBAAA,IAAI,8CAAK,MAAT,IAAI,CAAO,CAAC;gBAClB,OAAO;YACT,CAAC;YAED,IAAI,uBAAA,IAAI,iCAAe,EAAE,KAArB,IAAI,EAAkB,WAAW,CAAC,KAAK,KAAK,EAAE,CAAC;gBACjD,MAAM,uBAAA,IAAI,2DAAkB,MAAtB,IAAI,EACR,IAAI,KAAK,CACP,GAAG,uBAAA,IAAI,wBAAM,sCAAsC,IAAA,wBAAS,EAAC,WAAW,CAAC,EAAE,CAC5E,CACF,CAAC;gBACF,OAAO;YACT,CAAC;YAED,uBAAA,IAAI,0BAAQ,CAAC,GAAG,CAAC,IAAA,4BAAiB,EAAC,WAAW,CAAC,CAAC,CAAC;QACnD,CAAC,EAAC;QA/CA,uBAAA,IAAI,oBAAS,IAAI,IAAI,IAAI,CAAC,WAAW,CAAC,IAAI,MAAA,CAAC;QAC3C,uBAAA,IAAI,qBAAU,KAAK,MAAA,CAAC;QACpB,uBAAA,IAAI,6BAAkB,aAAa,MAAA,CAAC;QACpC,MAAM,CAAC,IAAI,CAAC,CAAC;IACf,CAAC;IAED;;;;;OAKG;IACO,eAAe;QACvB,IAAI,uBAAA,IAAI,yCAAuB,EAAE,CAAC;YAChC,MAAM,IAAI,KAAK,CACb,GAAG,uBAAA,IAAI,wBAAM,+CAA+C,CAC7D,CAAC;QACJ,CAAC;QACD,uBAAA,IAAI,qCAA0B,IAAI,MAAA,CAAC;QACnC,OAAO,uBAAA,IAAI,gCAAc,CAAC,IAAI,CAAC,IAAI,CAAC,CAAC;IACvC,CAAC;IAsDD;;;;OAIG;IACH;IA9BA;;;;OAIG;IACH,KAAK,uCAAmB,KAAY;QAClC,IAAI,CAAC,uBAAA,IAAI,0BAAQ,CAAC,eAAe,EAAE,EAAE,CAAC;YACpC,uBAAA,IAAI,0BAAQ,CAAC,GAAG,CAAC,KAAK,CAAC,CAAC;QAC1B,CAAC;QACD,MAAM,uBAAA,IAAI,8CAAK,MAAT,IAAI,EAAM,KAAK,CAAC,CAAC;IACzB,CAAC;IAED;;;;;OAKG;IACH,KAAK,0BAAM,KAAa;QACtB,uBAAA,IAAI,0BAAQ,CAAC,GAAG,CAAC,KAAK,CAAC,CAAC;QACxB,MAAM,MAAM,GAAG,uBAAA,IAAI,yBAAO,EAAE,KAAb,IAAI,EAAU,KAAK,CAAC,CAAC;QACpC,uBAAA,IAAI,qBAAU,SAAS,MAAA,CAAC;QACxB,MAAM,MAAM,CAAC;IACf,CAAC,EAOA,MAAM,CAAC,aAAa,EAAC;QACpB,OAAO,IAAI,CAAC;IACd,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,IAAI;QACR,OAAO,uBAAA,IAAI,0BAAQ,CAAC,GAAG,EAAE,CAAC;IAC5B,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,MAAM;QACV,MAAM,uBAAA,IAAI,8CAAK,MAAT,IAAI,CAAO,CAAC;QAClB,OAAO,IAAA,yBAAc,GAAE,CAAC;IAC1B,CAAC;IAED;;;;;;OAMG;IACH,KAAK,CAAC,KAAK,CAAC,KAAY;QACtB,MAAM,uBAAA,IAAI,8CAAK,MAAT,IAAI,EAAM,KAAK,CAAC,CAAC;QACvB,OAAO,IAAA,yBAAc,GAAE,CAAC;IAC1B,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,GAAG,CAAC,KAAa;QACrB,OAAO,KAAK,CAAC,CAAC,CAAC,IAAI,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC,CAAC,CAAC,IAAI,CAAC,MAAM,EAAE,CAAC;IACnD,CAAC;CACF;AApJD,gCAoJC;AACD,MAAM,CAAC,UAAU,CAAC,CAAC;AAYnB;;GAEG;AACH,MAAa,UAAU;IASrB;;;;;;;;OAQG;IACH,YAAY,EAAE,IAAI,EAAE,UAAU,EAAE,KAAK,EAAyB;;QAjB9D,6BAAmB,KAAK,EAAC;QAEhB,2BAAgB,YAAY,EAAC;QAE7B,yCAA6B;QAEtC,oCAA2B;QAYzB,uBAAA,IAAI,oBAAS,IAAI,IAAI,IAAI,CAAC,WAAW,CAAC,IAAI,MAAA,CAAC;QAC3C,uBAAA,IAAI,0BAAe,UAAU,MAAA,CAAC;QAC9B,uBAAA,IAAI,qBAAU,KAAK,MAAA,CAAC;QACpB,MAAM,CAAC,IAAI,CAAC,CAAC;IACf,CAAC;IA0DD;;;;OAIG;IACH;IA7DA;;;;;;;;;;;OAWG;IACH,KAAK,+BACH,KAAsB,EACtB,SAAS,GAAG,KAAK;QAEjB,IAAI,CAAC;YACH,MAAM,uBAAA,IAAI,8BAAY,MAAhB,IAAI,EAAa,IAAA,kBAAO,EAAC,KAAK,CAAC,CAAC,CAAC;YACvC,OAAO,KAAK,KAAK,2BAAgB,IAAI,KAAK,YAAY,KAAK;gBACzD,CAAC,CAAC,IAAA,yBAAc,GAAE;gBAClB,CAAC,CAAC,IAAA,4BAAiB,EAAC,SAAS,CAAC,CAAC;QACnC,CAAC;QAAC,OAAO,KAAK,EAAE,CAAC;YACf,IAAI,SAAS,EAAE,CAAC;gBACd,yFAAyF;gBACzF,+EAA+E;gBAC/E,MAAM,oBAAoB,GAAG,IAAI,KAAK,CACpC,GAAG,uBAAA,IAAI,wBAAM,0CAA0C,EACvD,EAAE,KAAK,EAAE,KAAK,EAAE,CACjB,CAAC;gBACF,MAAM,uBAAA,IAAI,8BAAY,MAAhB,IAAI,EAAa,IAAA,gCAAqB,EAAC,oBAAoB,CAAC,CAAC,CAAC;gBACpE,MAAM,oBAAoB,CAAC;YAC7B,CAAC;iBAAM,CAAC;gBACN,MAAM,uBAAA,IAAI,gDAAO,MAAX,IAAI;gBACR,sEAAsE;gBACtE,KAAK,YAAY,KAAK,CAAC,CAAC,CAAC,KAAK,CAAC,CAAC,CAAC,IAAI,KAAK,CAAC,MAAM,CAAC,KAAK,CAAC,CAAC,EACzD,IAAI,CACL,CAAC;gBACF,MAAM,IAAI,KAAK,CAAC,GAAG,uBAAA,IAAI,wBAAM,iCAAiC,EAAE;oBAC9D,KAAK,EAAE,KAAK;iBACb,CAAC,CAAC;YACL,CAAC;QACH,CAAC;IACH,CAAC;IAED;;;;OAIG;IACH,KAAK,0BAAM,KAAa;QACtB,uBAAA,IAAI,sBAAW,IAAI,MAAA,CAAC;QACpB,MAAM,MAAM,GAAG,uBAAA,IAAI,yBAAO,EAAE,KAAb,IAAI,EAAU,KAAK,CAAC,CAAC;QACpC,uBAAA,IAAI,qBAAU,SAAS,MAAA,CAAC;QACxB,MAAM,MAAM,CAAC;IACf,CAAC,EAOA,MAAM,CAAC,aAAa,EAAC;QACpB,OAAO,IAAI,CAAC;IACd,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,IAAI,CAAC,KAAY;QACrB,IAAI,uBAAA,IAAI,0BAAQ,EAAE,CAAC;YACjB,OAAO,IAAA,yBAAc,GAAE,CAAC;QAC1B,CAAC;QACD,OAAO,uBAAA,IAAI,mDAAU,MAAd,IAAI,EAAW,KAAK,CAAC,CAAC;IAC/B,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,MAAM;QACV,IAAI,CAAC,uBAAA,IAAI,0BAAQ,EAAE,CAAC;YAClB,MAAM,uBAAA,IAAI,8BAAY,MAAhB,IAAI,EAAa,IAAA,+BAAoB,GAAE,CAAC,CAAC;YAC/C,MAAM,uBAAA,IAAI,8CAAK,MAAT,IAAI,CAAO,CAAC;QACpB,CAAC;QACD,OAAO,IAAA,yBAAc,GAAE,CAAC;IAC1B,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,KAAK,CAAC,KAAY;QACtB,IAAI,CAAC,uBAAA,IAAI,0BAAQ,EAAE,CAAC;YAClB,MAAM,uBAAA,IAAI,gDAAO,MAAX,IAAI,EAAQ,KAAK,CAAC,CAAC;QAC3B,CAAC;QACD,OAAO,IAAA,yBAAc,GAAE,CAAC;IAC1B,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,GAAG,CAAC,KAAa;QACrB,OAAO,KAAK,CAAC,CAAC,CAAC,IAAI,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC,CAAC,CAAC,IAAI,CAAC,MAAM,EAAE,CAAC;IACnD,CAAC;CAqBF;AA9JD,gCA8JC;;AAnBC;;;;;;;;GAQG;AACH,KAAK,4BACH,KAAY,EACZ,SAAS,GAAG,KAAK;IAEjB,MAAM,MAAM,GAAG,uBAAA,IAAI,mDAAU,MAAd,IAAI,EAAW,KAAK,EAAE,SAAS,CAAC,CAAC;IAChD,IAAI,CAAC,uBAAA,IAAI,0BAAQ,EAAE,CAAC;QAClB,MAAM,uBAAA,IAAI,8CAAK,MAAT,IAAI,EAAM,KAAK,CAAC,CAAC;IACzB,CAAC;IACD,OAAO,MAAM,CAAC;AAChB,CAAC;AAEH,MAAM,CAAC,UAAU,CAAC,CAAC","sourcesContent":["import { makePromiseKit } from '@endo/promise-kit';\nimport type { Reader, Writer } from '@endo/stream';\nimport { stringify } from '@metamask/kernel-utils';\nimport type { PromiseCallbacks } from '@metamask/kernel-utils';\n\nimport type { Dispatchable, Writable } from './utils.ts';\nimport {\n  makeDoneResult,\n  makePendingResult,\n  makeStreamDoneSignal,\n  makeStreamErrorSignal,\n  marshal,\n  StreamDoneSymbol,\n  unmarshal,\n} from './utils.ts';\n\nconst makeStreamBuffer = <\n  Value extends IteratorResult<unknown, undefined>,\n  // eslint-disable-next-line @typescript-eslint/explicit-function-return-type\n>() => {\n  const inputBuffer: (Value | Error)[] = [];\n  const outputBuffer: PromiseCallbacks[] = [];\n  let done = false;\n\n  return {\n    /**\n     * Flushes pending reads with a value or error, and causes subsequent writes to be ignored.\n     * Subsequent reads will exhaust any puts, then return the error (if any), and finally a `done` result.\n     * Idempotent.\n     *\n     * @param error - The error to end the stream with. A `done` result is used if not provided.\n     */\n    end: (error?: Error): void => {\n      if (done) {\n        return;\n      }\n      done = true;\n\n      for (const { resolve, reject } of outputBuffer) {\n        error ? reject(error) : resolve(makeDoneResult() as Value);\n      }\n      outputBuffer.length = 0;\n    },\n\n    hasPendingReads(): boolean {\n      return outputBuffer.length > 0;\n    },\n\n    /**\n     * Puts a value or error into the buffer.\n     *\n     * @see `end()` for behavior when the stream ends.\n     * @param value - The value or error to put.\n     */\n    put(value: Value | Error): void {\n      if (done) {\n        return;\n      }\n\n      if (outputBuffer.length > 0) {\n        const { resolve } = outputBuffer.shift() as PromiseCallbacks;\n        resolve(value);\n        return;\n      }\n      inputBuffer.push(value);\n    },\n\n    async get(): Promise<Value> {\n      if (inputBuffer.length > 0) {\n        const value = inputBuffer.shift() as Value;\n        return value instanceof Error\n          ? Promise.reject(value)\n          : Promise.resolve(value);\n      }\n\n      if (done) {\n        return makeDoneResult() as Value;\n      }\n\n      const { promise, resolve, reject } = makePromiseKit<Value>();\n      outputBuffer.push({\n        resolve: resolve as (value: unknown) => void,\n        reject,\n      });\n      return promise;\n    },\n  };\n};\nharden(makeStreamBuffer);\n\n/**\n * A function that is called when a stream ends. Useful for cleanup, such as closing a\n * message port.\n */\nexport type OnEnd = (error?: Error) => void | Promise<void>;\n\n/**\n * A function that validates input to a readable stream.\n */\nexport type ValidateInput<Read> = (input: unknown) => input is Read;\n\n/**\n * A function that receives input from a transport mechanism to a readable stream.\n * Validates that the input is an {@link IteratorResult}, and throws if it is not.\n */\nexport type ReceiveInput = (input: unknown) => Promise<void>;\n\nexport type BaseReaderArgs<Read> = {\n  name?: string | undefined;\n  onEnd?: OnEnd | undefined;\n  validateInput?: ValidateInput<Read> | undefined;\n};\n\n/**\n * The base of a readable async iterator stream.\n *\n * Subclasses must forward input received from the transport mechanism via the function\n * returned by `getReceiveInput()`. Any cleanup required by subclasses should be performed\n * in a callback passed to `setOnEnd()`.\n *\n * The result of any value received before the stream ends is guaranteed to be observable\n * by the consumer.\n */\nexport class BaseReader<Read> implements Reader<Read> {\n  /**\n   * A buffer for managing backpressure (writes > reads) and \"suction\" (reads > writes) for a stream.\n   * Modeled on `AsyncQueue` from `@endo/stream`, but with arrays under the hood instead of a promise chain.\n   */\n  readonly #buffer = makeStreamBuffer<IteratorResult<Read, undefined>>();\n\n  readonly #name: string;\n\n  readonly #validateInput?: ValidateInput<Read> | undefined;\n\n  #onEnd?: OnEnd | undefined;\n\n  #didExposeReceiveInput: boolean = false;\n\n  /**\n   * Constructs a {@link BaseReader}.\n   *\n   * @param options - Options bag for configuring the reader.\n   * @param options.name - The name of the stream, for logging purposes. Defaults to the class name.\n   * @param options.onEnd - A function that is called when the stream ends. For any cleanup that\n   * should happen when the stream ends, such as closing a message port.\n   * @param options.validateInput - A function that validates input from the transport.\n   */\n  constructor({ name, onEnd, validateInput }: BaseReaderArgs<Read>) {\n    this.#name = name ?? this.constructor.name;\n    this.#onEnd = onEnd;\n    this.#validateInput = validateInput;\n    harden(this);\n  }\n\n  /**\n   * Returns the `receiveInput()` method, which is used to receive input from the stream.\n   * Attempting to call this method more than once will throw an error.\n   *\n   * @returns The `receiveInput()` method.\n   */\n  protected getReceiveInput(): ReceiveInput {\n    if (this.#didExposeReceiveInput) {\n      throw new Error(\n        `${this.#name} received multiple calls to getReceiveInput()`,\n      );\n    }\n    this.#didExposeReceiveInput = true;\n    return this.#receiveInput.bind(this);\n  }\n\n  readonly #receiveInput = async (input: unknown): Promise<void> => {\n    // eslint-disable-next-line @typescript-eslint/await-thenable\n    await null;\n\n    const unmarshaled = unmarshal(input);\n    if (unmarshaled instanceof Error) {\n      await this.#handleInputError(unmarshaled);\n      return;\n    }\n\n    if (unmarshaled === StreamDoneSymbol) {\n      await this.#end();\n      return;\n    }\n\n    if (this.#validateInput?.(unmarshaled) === false) {\n      await this.#handleInputError(\n        new Error(\n          `${this.#name}: Message failed type validation:\\n${stringify(unmarshaled)}`,\n        ),\n      );\n      return;\n    }\n\n    this.#buffer.put(makePendingResult(unmarshaled));\n  };\n\n  /**\n   * Handles an input error by putting it into the buffer and ending the stream.\n   *\n   * @param error - The error to handle.\n   */\n  async #handleInputError(error: Error): Promise<void> {\n    if (!this.#buffer.hasPendingReads()) {\n      this.#buffer.put(error);\n    }\n    await this.#end(error);\n  }\n\n  /**\n   * Ends the stream. Calls and then unsets the `#onEnd` method.\n   * Idempotent.\n   *\n   * @param error - The error to end the stream with. A `done` result is used if not provided.\n   */\n  async #end(error?: Error): Promise<void> {\n    this.#buffer.end(error);\n    const onEndP = this.#onEnd?.(error);\n    this.#onEnd = undefined;\n    await onEndP;\n  }\n\n  /**\n   * Returns the async iterator for this stream.\n   *\n   * @returns This stream as an async iterator.\n   */\n  [Symbol.asyncIterator](): typeof this {\n    return this;\n  }\n\n  /**\n   * Reads the next message from the transport.\n   *\n   * @returns The next message from the transport.\n   */\n  async next(): Promise<IteratorResult<Read, undefined>> {\n    return this.#buffer.get();\n  }\n\n  /**\n   * Closes the underlying transport and returns. Any unread messages will be lost.\n   *\n   * @returns The final result for this stream.\n   */\n  async return(): Promise<IteratorResult<Read, undefined>> {\n    await this.#end();\n    return makeDoneResult();\n  }\n\n  /**\n   * Rejects all pending reads with the specified error, closes the underlying transport,\n   * and returns.\n   *\n   * @param error - The error to reject pending reads with.\n   * @returns The final result for this stream.\n   */\n  async throw(error: Error): Promise<IteratorResult<Read, undefined>> {\n    await this.#end(error);\n    return makeDoneResult();\n  }\n\n  /**\n   * Closes the stream. Syntactic sugar for `return()` or `throw(error)`. Idempotent.\n   *\n   * @param error - The error to close the stream with.\n   * @returns The final result for this stream.\n   */\n  async end(error?: Error): Promise<IteratorResult<Read, undefined>> {\n    return error ? this.throw(error) : this.return();\n  }\n}\nharden(BaseReader);\n\nexport type Dispatch<Yield> = (\n  value: Dispatchable<Yield>,\n) => void | Promise<void>;\n\nexport type BaseWriterArgs<Write> = {\n  onDispatch: Dispatch<Write>;\n  name?: string | undefined;\n  onEnd?: OnEnd | undefined;\n};\n\n/**\n * The base of a writable async iterator stream.\n */\nexport class BaseWriter<Write> implements Writer<Write> {\n  #isDone: boolean = false;\n\n  readonly #name: string = 'BaseWriter';\n\n  readonly #onDispatch: Dispatch<Write>;\n\n  #onEnd?: OnEnd | undefined;\n\n  /**\n   * Constructs a {@link BaseWriter}.\n   *\n   * @param options - Options bag for configuring the writer.\n   * @param options.onDispatch - A function that dispatches messages over the underlying transport mechanism.\n   * @param options.name - The name of the stream, for logging purposes. Defaults to the class name.\n   * @param options.onEnd - A function that is called when the stream ends. For any cleanup that\n   * should happen when the stream ends, such as closing a message port.\n   */\n  constructor({ name, onDispatch, onEnd }: BaseWriterArgs<Write>) {\n    this.#name = name ?? this.constructor.name;\n    this.#onDispatch = onDispatch;\n    this.#onEnd = onEnd;\n    harden(this);\n  }\n\n  /**\n   * Dispatches the value, via the dispatch function registered in the constructor.\n   * If dispatching fails, calls `#throw()`, and is therefore mutually recursive with\n   * that method. For this reason, includes a flag indicating past failure to dispatch\n   * a value, which is used to avoid infinite recursion. If dispatching succeeds, returns a\n   * `{ done: true }` result if the value was an {@link Error} or itself a `done` result,\n   * otherwise returns `{ done: false }`.\n   *\n   * @param value - The value to dispatch.\n   * @param hasFailed - Whether dispatching has failed previously.\n   * @returns The result of dispatching the value.\n   */\n  async #dispatch(\n    value: Writable<Write>,\n    hasFailed = false,\n  ): Promise<IteratorResult<undefined, undefined>> {\n    try {\n      await this.#onDispatch(marshal(value));\n      return value === StreamDoneSymbol || value instanceof Error\n        ? makeDoneResult()\n        : makePendingResult(undefined);\n    } catch (error) {\n      if (hasFailed) {\n        // Break out of repeated failure to dispatch an error. It is unclear how this would occur\n        // in practice, but it's the kind of failure mode where it's better to be sure.\n        const repeatedFailureError = new Error(\n          `${this.#name} experienced repeated dispatch failures.`,\n          { cause: error },\n        );\n        await this.#onDispatch(makeStreamErrorSignal(repeatedFailureError));\n        throw repeatedFailureError;\n      } else {\n        await this.#throw(\n          /* istanbul ignore next: The ternary is mostly to please TypeScript */\n          error instanceof Error ? error : new Error(String(error)),\n          true,\n        );\n        throw new Error(`${this.#name} experienced a dispatch failure`, {\n          cause: error,\n        });\n      }\n    }\n  }\n\n  /**\n   * Ends the stream and calls the onEnd callback. Idempotent.\n   *\n   * @param error - The error to end the stream with.\n   */\n  async #end(error?: Error): Promise<void> {\n    this.#isDone = true;\n    const onEndP = this.#onEnd?.(error);\n    this.#onEnd = undefined;\n    await onEndP;\n  }\n\n  /**\n   * Returns the async iterator for this stream.\n   *\n   * @returns This stream as an async iterator.\n   */\n  [Symbol.asyncIterator](): typeof this {\n    return this;\n  }\n\n  /**\n   * Writes the next message to the transport.\n   *\n   * @param value - The next message to write to the transport.\n   * @returns The result of writing the message.\n   */\n  async next(value: Write): Promise<IteratorResult<undefined, undefined>> {\n    if (this.#isDone) {\n      return makeDoneResult();\n    }\n    return this.#dispatch(value);\n  }\n\n  /**\n   * Closes the underlying transport and returns. Idempotent.\n   *\n   * @returns The final result for this stream.\n   */\n  async return(): Promise<IteratorResult<undefined, undefined>> {\n    if (!this.#isDone) {\n      await this.#onDispatch(makeStreamDoneSignal());\n      await this.#end();\n    }\n    return makeDoneResult();\n  }\n\n  /**\n   * Forwards the error to the transport and closes this stream. Idempotent.\n   *\n   * @param error - The error to forward to the transport.\n   * @returns The final result for this stream.\n   */\n  async throw(error: Error): Promise<IteratorResult<undefined, undefined>> {\n    if (!this.#isDone) {\n      await this.#throw(error);\n    }\n    return makeDoneResult();\n  }\n\n  /**\n   * Closes the stream. Syntactic sugar for `return()` or `throw(error)`. Idempotent.\n   *\n   * @param error - The error to close the stream with.\n   * @returns The final result for this stream.\n   */\n  async end(error?: Error): Promise<IteratorResult<undefined, undefined>> {\n    return error ? this.throw(error) : this.return();\n  }\n\n  /**\n   * Dispatches the error and calls `#end()`. Mutually recursive with `dispatch()`.\n   * For this reason, includes a flag indicating past failure, so that `dispatch()`\n   * can avoid infinite recursion. See `dispatch()` for more details.\n   *\n   * @param error - The error to forward.\n   * @param hasFailed - Whether dispatching has failed previously.\n   * @returns The final result for this stream.\n   */\n  async #throw(\n    error: Error,\n    hasFailed = false,\n  ): Promise<IteratorResult<undefined, undefined>> {\n    const result = this.#dispatch(error, hasFailed);\n    if (!this.#isDone) {\n      await this.#end(error);\n    }\n    return result;\n  }\n}\nharden(BaseWriter);\n"]}