{"version":3,"file":"NodeWorkerStream.cjs","sourceRoot":"","sources":["../../src/node/NodeWorkerStream.ts"],"names":[],"mappings":";AAAA;;GAEG;;;AAEH,iEAGgC;AAMhC,qDAA0D;AAU1D;;;;;;GAMG;AACH,MAAa,gBAAuB,SAAQ,0BAAgB;IAC1D;;;;;;;OAOG;IACH,YACE,IAAc,EACd,EAAE,aAAa,EAAE,KAAK,KAA2B,EAAE;QAEnD,KAAK,CAAC;YACJ,aAAa;YACb,KAAK,EAAE,KAAK,IAAI,EAAE,CAAC,MAAM,KAAK,EAAE,EAAE;SACnC,CAAC,CAAC;QAEH,MAAM,YAAY,GAAG,KAAK,CAAC,eAAe,EAAE,CAAC;QAC7C,IAAI,CAAC,EAAE,CAAC,SAAS,EAAE,CAAC,IAAI,EAAE,EAAE;YAC1B,YAAY,CAAC,IAAI,CAAC,CAAC,KAAK,CAAC,KAAK,EAAE,KAAK,EAAE,EAAE,CAAC,IAAI,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC,CAAC;QAC/D,CAAC,CAAC,CAAC;QACH,MAAM,CAAC,IAAI,CAAC,CAAC;IACf,CAAC;CACF;AAxBD,4CAwBC;AACD,MAAM,CAAC,gBAAgB,CAAC,CAAC;AAEzB;;;;;;GAMG;AACH,MAAa,gBAAwB,SAAQ,0BAAiB;IAC5D;;;;;;;OAOG;IACH,YACE,IAAc,EACd,EAAE,IAAI,EAAE,KAAK,KAAgD,EAAE;QAE/D,KAAK,CAAC;YACJ,IAAI;YACJ,UAAU,EAAE,CAAC,KAA0B,EAAE,EAAE,CAAC,IAAI,CAAC,WAAW,CAAC,KAAK,CAAC;YACnE,KAAK,EAAE,KAAK,IAAI,EAAE;gBAChB,MAAM,KAAK,EAAE,EAAE,CAAC;YAClB,CAAC;SACF,CAAC,CAAC;QACH,MAAM,CAAC,IAAI,CAAC,CAAC;IACf,CAAC;CACF;AAtBD,4CAsBC;AACD,MAAM,CAAC,gBAAgB,CAAC,CAAC;AAEzB;;GAEG;AACH,MAAa,sBAGX,SAAQ,sCAKT;IACC;;;;;OAKG;IACH,YAAY,IAAc,EAAE,aAAmC;QAC7D,IAAI,MAA+B,CAAC,CAAC,mCAAmC;QACxE,MAAM,MAAM,GAAG,IAAI,gBAAgB,CAAO,IAAI,EAAE;YAC9C,IAAI,EAAE,wBAAwB;YAC9B,aAAa,EAAE,IAAA,oDAA8B,EAAC,aAAa,CAAC;YAC5D,KAAK,EAAE,KAAK,IAAI,EAAE;gBAChB,MAAM,MAAM,CAAC,MAAM,EAAE,CAAC;YACxB,CAAC;SACF,CAAC,CAAC;QACH,MAAM,GAAG,IAAI,gBAAgB,CAAQ,IAAI,EAAE;YACzC,IAAI,EAAE,wBAAwB;YAC9B,KAAK,EAAE,KAAK,IAAI,EAAE;gBAChB,MAAM,MAAM,CAAC,MAAM,EAAE,CAAC;YACxB,CAAC;SACF,CAAC,CAAC;QACH,KAAK,CAAC,MAAM,EAAE,MAAM,CAAC,CAAC;IACxB,CAAC;IAED;;;;;;OAMG;IACH,MAAM,CAAC,KAAK,CAAC,IAAI,CACf,IAAc,EACd,aAAmC;QAEnC,MAAM,MAAM,GAAG,IAAI,sBAAsB,CAAc,IAAI,EAAE,aAAa,CAAC,CAAC;QAC5E,MAAM,MAAM,CAAC,WAAW,EAAE,CAAC;QAC3B,OAAO,MAAM,CAAC;IAChB,CAAC;CACF;AAhDD,wDAgDC;AACD,MAAM,CAAC,sBAAsB,CAAC,CAAC","sourcesContent":["/**\n * @module Node Worker streams\n */\n\nimport {\n  BaseDuplexStream,\n  makeDuplexStreamInputValidator,\n} from '../BaseDuplexStream.ts';\nimport type {\n  BaseReaderArgs,\n  BaseWriterArgs,\n  ValidateInput,\n} from '../BaseStream.ts';\nimport { BaseReader, BaseWriter } from '../BaseStream.ts';\nimport type { Dispatchable } from '../utils.ts';\n\nexport type OnMessage = (message: unknown) => void;\n\nexport type NodePort = {\n  on: (event: 'message', listener: OnMessage) => void;\n  postMessage: (message: unknown) => void;\n};\n\n/**\n * A readable stream over a {@link NodePort}.\n *\n * @see\n * - {@link NodeWorkerWriter} for the corresponding writable stream.\n * - The module-level documentation for more details.\n */\nexport class NodeWorkerReader<Read> extends BaseReader<Read> {\n  /**\n   * Constructs a new {@link NodeWorkerReader}.\n   *\n   * @param port - The node worker port to read from.\n   * @param options - Options bag for configuring the reader.\n   * @param options.validateInput - A function that validates input from the transport.\n   * @param options.onEnd - A function that is called when the stream ends.\n   */\n  constructor(\n    port: NodePort,\n    { validateInput, onEnd }: BaseReaderArgs<Read> = {},\n  ) {\n    super({\n      validateInput,\n      onEnd: async () => await onEnd?.(),\n    });\n\n    const receiveInput = super.getReceiveInput();\n    port.on('message', (data) => {\n      receiveInput(data).catch(async (error) => this.throw(error));\n    });\n    harden(this);\n  }\n}\nharden(NodeWorkerReader);\n\n/**\n * A writable stream over a {@link NodeWorker}.\n *\n * @see\n * - {@link NodeWorkerReader} for the corresponding readable stream.\n * - The module-level documentation for more details.\n */\nexport class NodeWorkerWriter<Write> extends BaseWriter<Write> {\n  /**\n   * Constructs a new {@link NodeWorkerWriter}.\n   *\n   * @param port - The node worker port to write to.\n   * @param options - Options bag for configuring the writer.\n   * @param options.name - The name of the stream, for logging purposes.\n   * @param options.onEnd - A function that is called when the stream ends.\n   */\n  constructor(\n    port: NodePort,\n    { name, onEnd }: Omit<BaseWriterArgs<Write>, 'onDispatch'> = {},\n  ) {\n    super({\n      name,\n      onDispatch: (value: Dispatchable<Write>) => port.postMessage(value),\n      onEnd: async () => {\n        await onEnd?.();\n      },\n    });\n    harden(this);\n  }\n}\nharden(NodeWorkerWriter);\n\n/**\n * A duplex stream over a Node worker port.\n */\nexport class NodeWorkerDuplexStream<\n  Read,\n  Write = Read,\n> extends BaseDuplexStream<\n  Read,\n  NodeWorkerReader<Read>,\n  Write,\n  NodeWorkerWriter<Write>\n> {\n  /**\n   * Constructs a new {@link NodeWorkerDuplexStream}.\n   *\n   * @param port - The node worker port for bidirectional communication.\n   * @param validateInput - A function that validates input from the transport.\n   */\n  constructor(port: NodePort, validateInput?: ValidateInput<Read>) {\n    let writer: NodeWorkerWriter<Write>; // eslint-disable-line prefer-const\n    const reader = new NodeWorkerReader<Read>(port, {\n      name: 'NodeWorkerDuplexStream',\n      validateInput: makeDuplexStreamInputValidator(validateInput),\n      onEnd: async () => {\n        await writer.return();\n      },\n    });\n    writer = new NodeWorkerWriter<Write>(port, {\n      name: 'NodeWorkerDuplexStream',\n      onEnd: async () => {\n        await reader.return();\n      },\n    });\n    super(reader, writer);\n  }\n\n  /**\n   * Creates and synchronizes a new {@link NodeWorkerDuplexStream}.\n   *\n   * @param port - The node worker port for bidirectional communication.\n   * @param validateInput - A function that validates input from the transport.\n   * @returns A synchronized duplex stream.\n   */\n  static async make<Read, Write = Read>(\n    port: NodePort,\n    validateInput?: ValidateInput<Read>,\n  ): Promise<NodeWorkerDuplexStream<Read, Write>> {\n    const stream = new NodeWorkerDuplexStream<Read, Write>(port, validateInput);\n    await stream.synchronize();\n    return stream;\n  }\n}\nharden(NodeWorkerDuplexStream);\n"]}