{"version":3,"file":"BaseDuplexStream.cjs","sourceRoot":"","sources":["../src/BaseDuplexStream.ts"],"names":[],"mappings":";;;;;;;;;;;;;;;AACA,mDAAmD;AAEnD,yDAAmD;AACnD,uDAA4D;AAI5D,0CAA4C;AAE/B,QAAA,oBAAoB,GAAG;IAClC,GAAG,EAAE,OAAO;IACZ,GAAG,EAAE,OAAO;CACJ,CAAC;AAEX,MAAM,SAAS,GAAG,IAAA,oBAAM,EAAC;IACvB,CAAC,4BAAoB,CAAC,GAAG,CAAC,EAAE,IAAA,qBAAO,EAAC,IAAI,CAAC;CAC1C,CAAC,CAAC;AAII,MAAM,KAAK,GAAG,CAAC,KAAc,EAA4B,EAAE,CAChE,IAAA,gBAAE,EAAC,KAAK,EAAE,SAAS,CAAC,CAAC;AADV,QAAA,KAAK,SACK;AAEhB,MAAM,OAAO,GAAG,GAAoB,EAAE,CAAC,CAAC;IAC7C,CAAC,4BAAoB,CAAC,GAAG,CAAC,EAAE,IAAI;CACjC,CAAC,CAAC;AAFU,QAAA,OAAO,WAEjB;AAEH,MAAM,SAAS,GAAG,IAAA,oBAAM,EAAC;IACvB,CAAC,4BAAoB,CAAC,GAAG,CAAC,EAAE,IAAA,qBAAO,EAAC,IAAI,CAAC;CAC1C,CAAC,CAAC;AAII,MAAM,KAAK,GAAG,CAAC,KAAc,EAA4B,EAAE,CAChE,IAAA,gBAAE,EAAC,KAAK,EAAE,SAAS,CAAC,CAAC;AADV,QAAA,KAAK,SACK;AAEhB,MAAM,OAAO,GAAG,GAAoB,EAAE,CAAC,CAAC;IAC7C,CAAC,4BAAoB,CAAC,GAAG,CAAC,EAAE,IAAI;CACjC,CAAC,CAAC;AAFU,QAAA,OAAO,WAEjB;AAII,MAAM,oBAAoB,GAAG,CAClC,KAAc,EACe,EAAE,CAAC,IAAA,aAAK,EAAC,KAAK,CAAC,IAAI,IAAA,aAAK,EAAC,KAAK,CAAC,CAAC;AAFlD,QAAA,oBAAoB,wBAE8B;AAE/D;;;;;;;;;;GAUG;AACI,MAAM,8BAA8B,GAAG,CAC5C,aAAmC,EACc,EAAE,CACnD,aAAa;IACb,CAAC,CAAC,KAAc,EAAiB,EAAE,CACjC,IAAA,4BAAoB,EAAC,KAAK,CAAC,IAAI,aAAa,CAAC,KAAK,CAAC,CAAC,CAAC;AAL5C,QAAA,8BAA8B,kCAKc;AAEzD,MAAM,qBAAqB,GAAG;IAC5B,IAAI,EAAE,CAAC;IACP,OAAO,EAAE,CAAC;IACV,QAAQ,EAAE,CAAC;IACX,MAAM,EAAE,CAAC;CACD,CAAC;AAKX,MAAM,OAAO,GAAG,CAAC,MAA6B,EAAW,EAAE,CACzD,MAAM,KAAK,qBAAqB,CAAC,QAAQ;IACzC,MAAM,KAAK,qBAAqB,CAAC,MAAM,CAAC;AAE1C;;;GAGG;AACH,MAAsB,gBAAgB;IA2CpC;;;;;OAKG;IACH,YAAY,MAAkB,EAAE,MAAmB;;QA1CnD;;WAEG;QACM,2CAAoB;QAE7B;;WAEG;QACM,2CAAqB;QAE9B;;;WAGG;QACH,oCAA6B,IAAA,4BAAc,GAAQ,EAAC;QAEpD;;WAEG;QACH,kDAAgD,qBAAqB,CAAC,IAAI,EAAC;QAwBzE,2EAA2E;QAC3E,4EAA4E;QAC5E,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC,KAAK,CAAC,GAAG,EAAE,CAAC,SAAS,CAAC,CAAC;QAE7C,yDAAyD;QACzD,IAAI,CAAC,IAAI,GAAG,KAAK,IAAI,EAAE;YACrB,IAAI,uBAAA,IAAI,+CAAuB,KAAK,qBAAqB,CAAC,QAAQ,EAAE,CAAC;gBACnE,OAAO,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC,IAAI,CAAC,KAAK,IAAI,EAAE,CAAC,MAAM,CAAC,IAAI,EAAE,CAAC,CAAC;YAC/D,CAAC;YAED,MAAM,MAAM,GAAG,MAAM,MAAM,CAAC,IAAI,EAAE,CAAC;YAEnC,kDAAkD;YAClD,IAAI,IAAA,aAAK,EAAC,MAAM,CAAC,KAAK,CAAC,EAAE,CAAC;gBACxB,uBAAA,IAAI,iFAA4B,MAAhC,IAAI,CAA8B,CAAC;gBACnC,uBAAA,IAAI,6EAAwB,MAA5B,IAAI,EAAyB,MAAM,CAAC,CAAC,KAAK,CAAC,CAAC,KAAK,EAAE,EAAE;oBACnD,uBAAA,IAAI,0EAAqB,MAAzB,IAAI,EAAsB,KAAK,CAAC,CAAC;gBACnC,CAAC,CAAC,CAAC;gBACH,OAAO,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC,IAAI,CAAC,KAAK,IAAI,EAAE,CAAC,MAAM,CAAC,IAAI,EAAE,CAAC,CAAC;YAC/D,CAAC;YACD,OAAO,MAAM,CAAC;QAChB,CAAC,CAAC;QAEF,IAAI,CAAC,KAAK,GAAG,KAAK,EAAE,KAAY,EAAE,EAAE,CAClC,uBAAA,IAAI,+CAAuB,KAAK,qBAAqB,CAAC,QAAQ;YAC5D,CAAC,CAAC,MAAM,CAAC,IAAI,CAAC,KAAK,CAAC;YACpB,CAAC,CAAC,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC,IAAI,CAAC,KAAK,IAAI,EAAE,CAAC,MAAM,CAAC,IAAI,CAAC,KAAK,CAAC,CAAC,CAAC;QAEjE,uBAAA,IAAI,4BAAW,MAAM,MAAA,CAAC;QACtB,uBAAA,IAAI,4BAAW,MAAM,MAAA,CAAC;QAEtB,MAAM,CAAC,IAAI,CAAC,CAAC;IACf,CAAC;IAWD;;;;;OAKG;IACH,KAAK,CAAC,WAAW;QACf,IAAI,uBAAA,IAAI,+CAAuB,KAAK,qBAAqB,CAAC,IAAI,EAAE,CAAC;YAC/D,OAAO,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC;QAC/B,CAAC;QAED,IAAI,CAAC;YACH,MAAM,uBAAA,IAAI,6EAAwB,MAA5B,IAAI,CAA0B,CAAC;QACvC,CAAC;QAAC,OAAO,KAAK,EAAE,CAAC;YACf,uBAAA,IAAI,0EAAqB,MAAzB,IAAI,EAAsB,KAAc,CAAC,CAAC;QAC5C,CAAC;QAED,OAAO,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC;IAC/B,CAAC;IAqFD;;;;OAIG;IACH;QAjHE,uBAAA,IAAI,2CAA0B,qBAAqB,CAAC,IAAI,MAAA,CAAC;QACzD,uBAAA,IAAI,6BAAY,IAAA,4BAAc,GAAQ,MAAA,CAAC;QACvC,uBAAA,IAAI,iCAAS,CAAC,OAAO,CAAC,KAAK,CAAC,GAAG,EAAE,CAAC,SAAS,CAAC,CAAC;IAC/C,CAAC;IAsBD;;;;;;;;;;;;;OAaG;IACH,KAAK,mDACH,cAAgD;QAEhD,uBAAA,IAAI,2CAA0B,qBAAqB,CAAC,OAAO,MAAA,CAAC;QAC5D,IAAI,WAAW,GAAG,KAAK,CAAC;QACxB,IAAI,OAAO,GAAG,KAAK,CAAC;QAEpB,MAAM,OAAO,GAAG,KAAK,IAAmB,EAAE;YACxC,OAAO,GAAG,IAAI,CAAC;YACf,kCAAkC;YAClC,MAAM,uBAAA,IAAI,gCAAQ,CAAC,IAAI,CAAC,IAAA,eAAO,GAAE,CAAC,CAAC;QACrC,CAAC,CAAC;QAEF,kCAAkC;QAClC,MAAM,uBAAA,IAAI,gCAAQ,CAAC,IAAI,CAAC,IAAA,eAAO,GAAE,CAAC,CAAC;QAEnC,IAAI,MAAM,GAAG,cAAc,IAAI,CAAC,MAAM,uBAAA,IAAI,gCAAQ,CAAC,IAAI,EAAE,CAAC,CAAC;QAC3D,OAAO,uBAAA,IAAI,+CAAuB,KAAK,qBAAqB,CAAC,OAAO,EAAE,CAAC;YACrE,IAAI,IAAA,aAAK,EAAC,MAAM,CAAC,KAAK,CAAC,EAAE,CAAC;gBACxB,IAAI,CAAC,OAAO,EAAE,CAAC;oBACb,MAAM,OAAO,EAAE,CAAC;gBAClB,CAAC;gBACD,uBAAA,IAAI,8EAAyB,MAA7B,IAAI,CAA2B,CAAC;YAClC,CAAC;iBAAM,IAAI,IAAA,aAAK,EAAC,MAAM,CAAC,KAAK,CAAC,EAAE,CAAC;gBAC/B,IAAI,WAAW,EAAE,CAAC;oBAChB,uBAAA,IAAI,0EAAqB,MAAzB,IAAI,EACF,IAAI,KAAK,CAAC,uDAAuD,CAAC,CACnE,CAAC;gBACJ,CAAC;qBAAM,CAAC;oBACN,WAAW,GAAG,IAAI,CAAC;oBACnB,MAAM,OAAO,EAAE,CAAC;oBAChB,MAAM,GAAG,MAAM,uBAAA,IAAI,gCAAQ,CAAC,IAAI,EAAE,CAAC;gBACrC,CAAC;YACH,CAAC;iBAAM,CAAC;gBACN,uBAAA,IAAI,0EAAqB,MAAzB,IAAI,EACF,IAAI,KAAK,CACP,uDAAuD,IAAA,wBAAS,EAAC,MAAM,CAAC,EAAE,CAC3E,CACF,CAAC;YACJ,CAAC;QACH,CAAC;IACH,CAAC;QAMC,IAAI,OAAO,CAAC,uBAAA,IAAI,+CAAuB,CAAC,EAAE,CAAC;YACzC,OAAO;QACT,CAAC;QAED,uBAAA,IAAI,2CAA0B,qBAAqB,CAAC,QAAQ,MAAA,CAAC;QAC7D,uBAAA,IAAI,iCAAS,CAAC,OAAO,EAAE,CAAC;IAC1B,CAAC,yFAOoB,KAAY;QAC/B,IAAI,OAAO,CAAC,uBAAA,IAAI,+CAAuB,CAAC,EAAE,CAAC;YACzC,OAAO;QACT,CAAC;QAED,uBAAA,IAAI,2CAA0B,qBAAqB,CAAC,MAAM,MAAA,CAAC;QAC3D,uBAAA,IAAI,iCAAS,CAAC,MAAM,CAAC,KAAK,CAAC,CAAC;IAC9B,CAAC,EAOA,MAAM,CAAC,aAAa,EAAC;QACpB,OAAO,IAAI,CAAC;IACd,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,KAAK,CAAC,OAA8C;QACxD,IAAI,KAAK,EAAE,MAAM,KAAK,IAAI,IAAI,EAAE,CAAC;YAC/B,MAAM,OAAO,CAAC,KAAK,CAAC,CAAC;QACvB,CAAC;IACH,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,IAAI,CAAQ,IAA+B;QAC/C,MAAM,IAAI,CAAC,KAAK,CAAC,KAAK,EAAE,KAAK,EAAE,EAAE;YAC/B,MAAM,IAAI,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC;QAC1B,CAAC,CAAC,CAAC;IACL,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,MAAM;QACV,uBAAA,IAAI,8EAAyB,MAA7B,IAAI,CAA2B,CAAC;QAChC,MAAM,OAAO,CAAC,GAAG,CAAC,CAAC,uBAAA,IAAI,gCAAQ,CAAC,MAAM,EAAE,EAAE,uBAAA,IAAI,gCAAQ,CAAC,MAAM,EAAE,CAAC,CAAC,CAAC;QAClE,OAAO,IAAA,yBAAc,GAAE,CAAC;IAC1B,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,KAAK,CAAC,KAAY;QACtB,uBAAA,IAAI,0EAAqB,MAAzB,IAAI,EAAsB,KAAK,CAAC,CAAC;QACjC,0DAA0D;QAC1D,MAAM,OAAO,CAAC,GAAG,CAAC,CAAC,uBAAA,IAAI,gCAAQ,CAAC,KAAK,CAAC,KAAK,CAAC,EAAE,uBAAA,IAAI,gCAAQ,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC,CAAC,CAAC;QAC1E,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;AApQD,4CAoQC;AACD,MAAM,CAAC,gBAAgB,CAAC,CAAC","sourcesContent":["import type { PromiseKit } from '@endo/promise-kit';\nimport { makePromiseKit } from '@endo/promise-kit';\nimport type { Reader } from '@endo/stream';\nimport { stringify } from '@metamask/kernel-utils';\nimport { is, literal, object } from '@metamask/superstruct';\nimport type { Infer } from '@metamask/superstruct';\n\nimport type { BaseReader, BaseWriter, ValidateInput } from './BaseStream.ts';\nimport { makeDoneResult } from './utils.ts';\n\nexport const DuplexStreamSentinel = {\n  Syn: '@@Syn',\n  Ack: '@@Ack',\n} as const;\n\nconst SynStruct = object({\n  [DuplexStreamSentinel.Syn]: literal(true),\n});\n\ntype DuplexStreamSyn = Infer<typeof SynStruct>;\n\nexport const isSyn = (value: unknown): value is DuplexStreamSyn =>\n  is(value, SynStruct);\n\nexport const makeSyn = (): DuplexStreamSyn => ({\n  [DuplexStreamSentinel.Syn]: true,\n});\n\nconst AckStruct = object({\n  [DuplexStreamSentinel.Ack]: literal(true),\n});\n\ntype DuplexStreamAck = Infer<typeof AckStruct>;\n\nexport const isAck = (value: unknown): value is DuplexStreamAck =>\n  is(value, AckStruct);\n\nexport const makeAck = (): DuplexStreamAck => ({\n  [DuplexStreamSentinel.Ack]: true,\n});\n\ntype DuplexStreamSignal = DuplexStreamSyn | DuplexStreamAck;\n\nexport const isDuplexStreamSignal = (\n  value: unknown,\n): value is DuplexStreamSignal => isSyn(value) || isAck(value);\n\n/**\n * Make a validator for input to a duplex stream. Constructor helper for concrete\n * duplex stream implementations.\n *\n * Validators passed in by consumers must be augmented such that errors aren't\n * thrown for {@link DuplexStreamSignal} values.\n *\n * @param validateInput - The validator for the stream's input type.\n * @returns A validator for the stream's input type, or `undefined` if no\n * validation is desired.\n */\nexport const makeDuplexStreamInputValidator = <Read>(\n  validateInput?: ValidateInput<Read>,\n): ((value: unknown) => value is Read) | undefined =>\n  validateInput &&\n  ((value: unknown): value is Read =>\n    isDuplexStreamSignal(value) || validateInput(value));\n\nconst SynchronizationStatus = {\n  Idle: 0,\n  Pending: 1,\n  Complete: 2,\n  Failed: 3,\n} as const;\n\ntype SynchronizationStatus =\n  (typeof SynchronizationStatus)[keyof typeof SynchronizationStatus];\n\nconst isEnded = (status: SynchronizationStatus): boolean =>\n  status === SynchronizationStatus.Complete ||\n  status === SynchronizationStatus.Failed;\n\n/**\n * The base of a duplex stream. Essentially a {@link BaseReader} with a `write()` method.\n * Backed up by separate {@link BaseReader} and {@link BaseWriter} instances under the hood.\n */\nexport abstract class BaseDuplexStream<\n  Read,\n  ReadStream extends BaseReader<Read>,\n  Write = Read,\n  WriteStream extends BaseWriter<Write> = BaseWriter<Write>,\n> implements Reader<Read>\n{\n  /**\n   * The underlying reader for the duplex stream.\n   */\n  readonly #reader: ReadStream;\n\n  /**\n   * The underlying writer for the duplex stream.\n   */\n  readonly #writer: WriteStream;\n\n  /**\n   * The promise for the synchronization of the stream with its remote\n   * counterpart.\n   */\n  #syncKit: PromiseKit<void> = makePromiseKit<void>();\n\n  /**\n   * Whether the stream is synchronized with its remote counterpart.\n   */\n  #synchronizationStatus: SynchronizationStatus = SynchronizationStatus.Idle;\n\n  /**\n   * Reads the next value from the stream.\n   *\n   * @returns The next value from the stream.\n   */\n  next: () => Promise<IteratorResult<Read, undefined>>;\n\n  /**\n   * Writes a value to the stream.\n   *\n   * @param value - The next value to write to the stream.\n   * @returns The result of writing the value.\n   */\n  write: (value: Write) => Promise<IteratorResult<undefined, undefined>>;\n\n  /**\n   * Constructs a new {@link BaseDuplexStream}.\n   *\n   * @param reader - The underlying reader for the duplex stream.\n   * @param writer - The underlying writer for the duplex stream.\n   */\n  constructor(reader: ReadStream, writer: WriteStream) {\n    // Set a catch handler to avoid unhandled rejection errors. The promise may\n    // reject before reads or writes occur, in which case there are no handlers.\n    this.#syncKit.promise.catch(() => undefined);\n\n    // Next and write only work if synchronization completes.\n    this.next = async () => {\n      if (this.#synchronizationStatus !== SynchronizationStatus.Complete) {\n        return this.#syncKit.promise.then(async () => reader.next());\n      }\n\n      const result = await reader.next();\n\n      // If we receive a SYN message, we re-synchronize.\n      if (isSyn(result.value)) {\n        this.#resetSynchronizationStatus();\n        this.#performSynchronization(result).catch((error) => {\n          this.#failSynchronization(error);\n        });\n        return this.#syncKit.promise.then(async () => reader.next());\n      }\n      return result;\n    };\n\n    this.write = async (value: Write) =>\n      this.#synchronizationStatus === SynchronizationStatus.Complete\n        ? writer.next(value)\n        : this.#syncKit.promise.then(async () => writer.next(value));\n\n    this.#reader = reader;\n    this.#writer = writer;\n\n    harden(this);\n  }\n\n  /**\n   * Resets the synchronization status to idle and creates a new promise kit.\n   */\n  #resetSynchronizationStatus(): void {\n    this.#synchronizationStatus = SynchronizationStatus.Idle;\n    this.#syncKit = makePromiseKit<void>();\n    this.#syncKit.promise.catch(() => undefined);\n  }\n\n  /**\n   * Synchronizes the duplex stream with its remote counterpart. Must be awaited\n   * before values can be read from or written to the stream. Idempotent.\n   *\n   * @returns A promise that resolves when the stream is synchronized.\n   */\n  async synchronize(): Promise<void> {\n    if (this.#synchronizationStatus !== SynchronizationStatus.Idle) {\n      return this.#syncKit.promise;\n    }\n\n    try {\n      await this.#performSynchronization();\n    } catch (error) {\n      this.#failSynchronization(error as Error);\n    }\n\n    return this.#syncKit.promise;\n  }\n\n  /**\n   * Performs the synchronization protocol.\n   *\n   * **ATTN:** The synchronization protocol requires sending values that do not\n   * conform to the read and write types of the stream. We do not currently have\n   * the type system to express this, so we just override TypeScript and do it\n   * anyway. This is far from ideal, but it works because (1) the streams always\n   * allow stream signal values at runtime, and (2) the special values cannot\n   * be observed by users of the stream.\n   *\n   * @param previousResult - The result of the previous `next()` call.\n   * This method will call `next()` on the reader if not provided. Used\n   * when the other side reinitializes synchronization.\n   */\n  async #performSynchronization(\n    previousResult?: IteratorResult<Read, undefined>,\n  ): Promise<void> {\n    this.#synchronizationStatus = SynchronizationStatus.Pending;\n    let receivedSyn = false;\n    let sentAck = false;\n\n    const sendAck = async (): Promise<void> => {\n      sentAck = true;\n      // @ts-expect-error See docstring.\n      await this.#writer.next(makeAck());\n    };\n\n    // @ts-expect-error See docstring.\n    await this.#writer.next(makeSyn());\n\n    let result = previousResult ?? (await this.#reader.next());\n    while (this.#synchronizationStatus === SynchronizationStatus.Pending) {\n      if (isAck(result.value)) {\n        if (!sentAck) {\n          await sendAck();\n        }\n        this.#completeSynchronization();\n      } else if (isSyn(result.value)) {\n        if (receivedSyn) {\n          this.#failSynchronization(\n            new Error('Received duplicate SYN message during synchronization'),\n          );\n        } else {\n          receivedSyn = true;\n          await sendAck();\n          result = await this.#reader.next();\n        }\n      } else {\n        this.#failSynchronization(\n          new Error(\n            `Received unexpected message during synchronization: ${stringify(result)}`,\n          ),\n        );\n      }\n    }\n  }\n\n  /**\n   * Completes the synchronization by resolving the sync promise. Idempotent.\n   */\n  #completeSynchronization(): void {\n    if (isEnded(this.#synchronizationStatus)) {\n      return;\n    }\n\n    this.#synchronizationStatus = SynchronizationStatus.Complete;\n    this.#syncKit.resolve();\n  }\n\n  /**\n   * Fails the synchronization by rejecting the sync promise with the given error. Idempotent.\n   *\n   * @param error - The error to reject the sync promise with.\n   */\n  #failSynchronization(error: Error): void {\n    if (isEnded(this.#synchronizationStatus)) {\n      return;\n    }\n\n    this.#synchronizationStatus = SynchronizationStatus.Failed;\n    this.#syncKit.reject(error);\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   * Drains the stream by passing each value to a handler function.\n   *\n   * @param handler - The function that will receive each value from the stream.\n   */\n  async drain(handler: (value: Read) => void | Promise<void>): Promise<void> {\n    for await (const value of this) {\n      await handler(value);\n    }\n  }\n\n  /**\n   * Pipes the stream to another duplex stream.\n   *\n   * @param sink - The duplex stream to pipe to.\n   */\n  async pipe<Read2>(sink: DuplexStream<Read2, Read>): Promise<void> {\n    await this.drain(async (value) => {\n      await sink.write(value);\n    });\n  }\n\n  /**\n   * Closes the stream. Idempotent.\n   *\n   * @returns The final result for this stream.\n   */\n  async return(): Promise<IteratorResult<Read, undefined>> {\n    this.#completeSynchronization();\n    await Promise.all([this.#writer.return(), this.#reader.return()]);\n    return makeDoneResult();\n  }\n\n  /**\n   * Closes the stream with an error. Idempotent.\n   *\n   * @param error - The error to close the stream with.\n   * @returns The final result for this stream.\n   */\n  async throw(error: Error): Promise<IteratorResult<Read, undefined>> {\n    this.#failSynchronization(error);\n    // eslint-disable-next-line promise/no-promise-in-callback\n    await Promise.all([this.#writer.throw(error), this.#reader.throw(error)]);\n    return makeDoneResult();\n  }\n\n  /**\n   * Closes the stream. Syntactic sugar for `throw(error)` or `return()`. 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(BaseDuplexStream);\n\n/**\n * A duplex stream. Essentially a {@link Reader} with a `write()` method.\n */\nexport type DuplexStream<Read, Write = Read> = Pick<\n  BaseDuplexStream<Read, BaseReader<Read>, Write, BaseWriter<Write>>,\n  'next' | 'write' | 'drain' | 'pipe' | 'return' | 'throw' | 'end'\n> & {\n  [Symbol.asyncIterator]: () => DuplexStream<Read, Write>;\n};\n"]}