{"version":3,"file":"split.cjs","sourceRoot":"","sources":["../src/split.ts"],"names":[],"mappings":";;;;;;;;;;;;;;AAgOA,sBAoCC;AApQD,yDAAmD;AAGnD,oDAA6C;AAG7C;;;;GAIG;AACH,MAAM,WAAkB,SAAQ,0BAAgB;IAC9C;;;;OAIG;IACH,gDAAgD;IAChD,YAAoB,IAA0B;QAC5C,KAAK,CAAC,IAAI,CAAC,CAAC;IACd,CAAC;IAED;;;;;OAKG;IACH,MAAM,CAAC,IAAI,CACT,IAA0B;QAE1B,MAAM,MAAM,GAAG,IAAI,WAAW,CAAO,IAAI,CAAC,CAAC;QAC3C,OAAO,CAAC,MAAM,EAAE,MAAM,CAAC,eAAe,EAAE,CAAU,CAAC;IACrD,CAAC;CACF;AACD,MAAM,CAAC,WAAW,CAAC,CAAC;AAEpB;;;;GAIG;AACH,MAAM,WAAW;IAOf;;;;;OAKG;IACH,YACE,MAAuC,EACvC,MAAyB;QAZlB,sCAAyC;QAEzC,sCAA2B;QAYlC,uBAAA,IAAI,uBAAW,MAAM,MAAA,CAAC;QACtB,uBAAA,IAAI,uBAAW,MAAM,MAAA,CAAC;QACtB,MAAM,CAAC,IAAI,CAAC,CAAC;IACf,CAAC;IAED;;;;;OAKG;IACH,MAAM,CAAC,IAAI,CACT,MAAuC;QAKvC,MAAM,CAAC,MAAM,EAAE,YAAY,CAAC,GAAG,WAAW,CAAC,IAAI,CAAO;YACpD,IAAI,EAAE,IAAI,CAAC,WAAW,CAAC,IAAI;SAC5B,CAAC,CAAC;QACH,MAAM,MAAM,GAAG,IAAI,WAAW,CAAC,MAAM,EAAE,MAAM,CAAC,CAAC;QAC/C,OAAO,EAAE,MAAM,EAAE,YAAY,EAAE,CAAC;IAClC,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,IAAI;QACR,OAAO,uBAAA,IAAI,2BAAQ,CAAC,IAAI,EAAE,CAAC;IAC7B,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,KAAK,CAAC,KAAY;QACtB,OAAO,uBAAA,IAAI,2BAAQ,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC;IACnC,CAAC;IAED;;;;OAIG;IACH,KAAK,CAAC,KAAK,CAAC,OAA8C;QACxD,IAAI,KAAK,EAAE,MAAM,KAAK,IAAI,uBAAA,IAAI,2BAAQ,EAAE,CAAC;YACvC,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,MAAM,uBAAA,IAAI,2BAAQ,CAAC,MAAM,EAAE,CAAC;QAC5B,OAAO,uBAAA,IAAI,2BAAQ,CAAC,MAAM,EAAE,CAAC;IAC/B,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,KAAK,CAAC,KAAY;QACtB,MAAM,uBAAA,IAAI,2BAAQ,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC;QAChC,OAAO,uBAAA,IAAI,2BAAQ,CAAC,KAAK,CAAC,KAAK,CAAC,CAAC;IACnC,CAAC;IAED;;;;;OAKG;IACH,KAAK,CAAC,GAAG,CAAC,KAAa;QACrB,MAAM,uBAAA,IAAI,2BAAQ,CAAC,GAAG,CAAC,KAAK,CAAC,CAAC;QAC9B,OAAO,uBAAA,IAAI,2BAAQ,CAAC,GAAG,CAAC,KAAK,CAAC,CAAC;IACjC,CAAC;IAED;;;;OAIG;IACH,4EAAC,MAAM,CAAC,aAAa,EAAC;QACpB,OAAO,IAAI,CAAC;IACd,CAAC;CACF;AACD,MAAM,CAAC,WAAW,CAAC,CAAC;AAmDpB;;;;;;;GAOG;AACH,SAAgB,KAAK,CACnB,YAAuC,EACvC,GAAG,UAAwC;IAE3C,MAAM,MAAM,GAAG,UAAU,CAAC,GAAG,CAC3B,CAAC,SAAS,EAAE,EAAE,CAAC,CAAC,SAAS,EAAE,WAAW,CAAC,IAAI,CAAC,YAAY,CAAC,CAAU,CACpE,CAAC;IAEF,mCAAmC;IACnC,KAAK,CAAC,KAAK,IAAI,EAAE;QACf,IAAI,KAAwB,CAAC;QAC7B,IAAI,CAAC;YACH,IAAI,KAAK,EAAE,MAAM,KAAK,IAAI,YAAY,EAAE,CAAC;gBACvC,IAAI,OAAO,GAAG,KAAK,CAAC;gBACpB,KAAK,MAAM,CAAC,SAAS,EAAE,EAAE,YAAY,EAAE,CAAC,IAAI,MAAM,EAAE,CAAC;oBACnD,IAAI,SAAS,CAAC,KAAK,CAAC,EAAE,CAAC;wBACrB,OAAO,GAAG,IAAI,CAAC;wBACf,MAAM,YAAY,CAAC,KAAK,CAAC,CAAC;wBAC1B,MAAM;oBACR,CAAC;gBACH,CAAC;gBAED,IAAI,CAAC,OAAO,EAAE,CAAC;oBACb,MAAM,IAAI,KAAK,CACb,4CAA4C,IAAA,wBAAS,EAAC,KAAK,CAAC,EAAE,CAC/D,CAAC;gBACJ,CAAC;YACH,CAAC;QACH,CAAC;QAAC,OAAO,WAAW,EAAE,CAAC;YACrB,KAAK,GAAG,WAAoB,CAAC;QAC/B,CAAC;QAED,MAAM,OAAO,CAAC,GAAG,CAAC,MAAM,CAAC,GAAG,CAAC,KAAK,EAAE,CAAC,EAAE,EAAE,MAAM,EAAE,CAAC,EAAE,EAAE,CAAC,MAAM,CAAC,GAAG,CAAC,KAAK,CAAC,CAAC,CAAC,CAAC;IAC7E,CAAC,CAAC,EAAE,CAAC;IAEL,OAAO,MAAM,CAAC,GAAG,CAAC,CAAC,CAAC,EAAE,EAAE,MAAM,EAAE,CAAC,EAAE,EAAE,CAAC,MAAM,CAAC,CAAC;AAChD,CAAC","sourcesContent":["import { stringify } from '@metamask/kernel-utils';\n\nimport type { DuplexStream } from './BaseDuplexStream.ts';\nimport { BaseReader } from './BaseStream.ts';\nimport type { BaseReaderArgs, ReceiveInput } from './BaseStream.ts';\n\n/**\n * A reader for use within {@link split} that reads from a reader and forwards\n * writes to a parent. The reader should output a subset of the parent stream's values based\n * on some predicate.\n */\nclass SplitReader<Read> extends BaseReader<Read> {\n  /**\n   * Constructs a new {@link SplitReader}.\n   *\n   * @param args - The arguments to pass to the base reader.\n   */\n  // eslint-disable-next-line no-restricted-syntax\n  private constructor(args: BaseReaderArgs<Read>) {\n    super(args);\n  }\n\n  /**\n   * Creates a new {@link SplitReader}.\n   *\n   * @param args - The arguments to pass to the base reader.\n   * @returns A new {@link SplitReader} and the receive input function.\n   */\n  static make<Read>(\n    args: BaseReaderArgs<Read>,\n  ): [SplitReader<Read>, ReceiveInput] {\n    const reader = new SplitReader<Read>(args);\n    return [reader, reader.getReceiveInput()] as const;\n  }\n}\nharden(SplitReader);\n\n/**\n * A {@link DuplexStream} for use within {@link split} that reads from a reader and forwards\n * writes to a parent. The reader should output a subset of the parent stream's values based\n * on some predicate.\n */\nclass SplitStream<ParentRead, Read extends ParentRead, Write>\n  implements DuplexStream<Read, Write>\n{\n  readonly #parent: DuplexStream<ParentRead, Write>;\n\n  readonly #reader: SplitReader<Read>;\n\n  /**\n   * Constructs a new {@link SplitStream}.\n   *\n   * @param parent - The parent stream to read from.\n   * @param reader - The reader to use to read from the parent stream.\n   */\n  constructor(\n    parent: DuplexStream<ParentRead, Write>,\n    reader: SplitReader<Read>,\n  ) {\n    this.#parent = parent;\n    this.#reader = reader;\n    harden(this);\n  }\n\n  /**\n   * Constructs a new {@link SplitStream}.\n   *\n   * @param parent - The parent stream to read from.\n   * @returns A new {@link SplitStream} and the receive input function.\n   */\n  static make<ParentRead, Read extends ParentRead, Write>(\n    parent: DuplexStream<ParentRead, Write>,\n  ): {\n    stream: SplitStream<ParentRead, Read, Write>;\n    receiveInput: ReceiveInput;\n  } {\n    const [reader, receiveInput] = SplitReader.make<Read>({\n      name: this.constructor.name,\n    });\n    const stream = new SplitStream(parent, reader);\n    return { stream, receiveInput };\n  }\n\n  /**\n   * Reads the next value from the stream.\n   *\n   * @returns The next value from the stream.\n   */\n  async next(): Promise<IteratorResult<Read, undefined>> {\n    return this.#reader.next();\n  }\n\n  /**\n   * Writes a value to the stream.\n   *\n   * @param value - The value to write to the stream.\n   * @returns The result of writing the value.\n   */\n  async write(value: Write): Promise<IteratorResult<undefined, undefined>> {\n    return this.#parent.write(value);\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.#reader) {\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    await this.#parent.return();\n    return this.#reader.return();\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    await this.#parent.throw(error);\n    return this.#reader.throw(error);\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    await this.#parent.end(error);\n    return this.#reader.end(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}\nharden(SplitStream);\n\n// There's no reason to do this but we leave it in for the sake of completeness.\nexport function split<Read, Write, ReadA extends Read>(\n  stream: DuplexStream<Read, Write>,\n  predicateA: (value: Read) => value is ReadA,\n): [DuplexStream<ReadA, Write>];\n\nexport function split<Read, Write, ReadA extends Read, ReadB extends Read>(\n  stream: DuplexStream<Read, Write>,\n  predicateA: (value: Read) => value is ReadA,\n  predicateB: (value: Read) => value is ReadB,\n): [DuplexStream<ReadA, Write>, DuplexStream<ReadB, Write>];\n\nexport function split<\n  Read,\n  Write,\n  ReadA extends Read,\n  ReadB extends Read,\n  ReadC extends Read,\n>(\n  stream: DuplexStream<Read, Write>,\n  predicateA: (value: Read) => value is ReadA,\n  predicateB: (value: Read) => value is ReadB,\n  predicateC: (value: Read) => value is ReadC,\n): [\n  DuplexStream<ReadA, Write>,\n  DuplexStream<ReadB, Write>,\n  DuplexStream<ReadC, Write>,\n];\n\nexport function split<\n  Read,\n  Write,\n  ReadA extends Read,\n  ReadB extends Read,\n  ReadC extends Read,\n  ReadD extends Read,\n>(\n  stream: DuplexStream<Read, Write>,\n  predicateA: (value: Read) => value is ReadA,\n  predicateB: (value: Read) => value is ReadB,\n  predicateC: (value: Read) => value is ReadC,\n  predicateD: (value: Read) => value is ReadD,\n): [\n  DuplexStream<ReadA, Write>,\n  DuplexStream<ReadB, Write>,\n  DuplexStream<ReadC, Write>,\n  DuplexStream<ReadD, Write>,\n];\n\n/**\n * Splits a stream into multiple streams based on a list of predicates.\n * Supports up to 4 predicates with type checking, and any number without!\n *\n * @param parentStream - The stream to split.\n * @param predicates - The predicates to use to split the stream.\n * @returns An array of \"splits\" of the parent stream.\n */\nexport function split<Read, Write>(\n  parentStream: DuplexStream<Read, Write>,\n  ...predicates: ((value: Read) => boolean)[]\n): DuplexStream<Read, Write>[] {\n  const splits = predicates.map(\n    (predicate) => [predicate, SplitStream.make(parentStream)] as const,\n  );\n\n  // eslint-disable-next-line no-void\n  void (async () => {\n    let error: Error | undefined;\n    try {\n      for await (const value of parentStream) {\n        let matched = false;\n        for (const [predicate, { receiveInput }] of splits) {\n          if (predicate(value)) {\n            matched = true;\n            await receiveInput(value);\n            break;\n          }\n        }\n\n        if (!matched) {\n          throw new Error(\n            `Failed to match any predicate for value: ${stringify(value)}`,\n          );\n        }\n      }\n    } catch (caughtError) {\n      error = caughtError as Error;\n    }\n\n    await Promise.all(splits.map(async ([, { stream }]) => stream.end(error)));\n  })();\n\n  return splits.map(([, { stream }]) => stream);\n}\n"]}