123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278
import { Channel, ChannelOptions, Receiver } from "./channel.ts";import {  ignoreAbortedError,  isPositiveSafeInteger,  makeAbortCtrl,  recordWithDefaults,} from "./internal/utils.ts";export * from "./subscribe.ts";
/** * Options for pipe operations. */export interface ChannelPipeOptions extends ChannelOptions {  /**   *  A safe integer representing the channel buffer size.   *  A `bufferSize` of `0` indicates a channel without any buffer.   *  A negative `bufferSize` indicates a channel with an endless buffer.   * @type {number}   */  bufferSize?: number;  /**   * If provided, the pipe operation will halt when the signal is triggered.   *   * @type {AbortSignal}   */  signal?: AbortSignal;}
export function map<T, TOut>(  fn: (val: T) => TOut | Promise<TOut>,  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<T>): Receiver<TOut> => {    const { signal, bufferSize, ...options } = pipeOpts ?? {};    const outChan = new Channel<TOut>(      bufferSize ?? target.bufferSize,      options,    );    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;        await outChan.send(await fn(res[0]), makeAbortCtrl(signal));      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("map", fn, err))      .finally(() => outChan.close());    return outChan;  };}
export function flat<TElem>(  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<Iterable<TElem> | AsyncIterable<TElem>>) => {    const { signal, bufferSize, ...options } = pipeOpts ?? {};    const outChan = new Channel<TElem>(      bufferSize ?? target.bufferSize,      options,    );    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;
        for await (const item of res[0]) {          await outChan.send(item, makeAbortCtrl(signal));        }      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("flat", err))      .finally(() => outChan.close());
    return outChan;  };}
export function flatMap<T, TOut>(  fn: (val: T) =>    | Iterable<TOut>    | AsyncIterable<TOut>    | Promise<Iterable<TOut>>    | Promise<AsyncIterable<TOut>>,  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<T>) => {    const { signal, bufferSize, ...options } = pipeOpts ?? {};    const outChan = new Channel<TOut>(      bufferSize ?? target.bufferSize,      options,    );    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;        for await (const item of await fn(res[0])) {          await outChan.send(item, makeAbortCtrl(signal));        }      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("flatMap", fn, err))      .finally(() => outChan.close());    return outChan;  };}
export function forEach<T>(  fn: (val: T) => unknown | Promise<unknown>,  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<T>) => {    const { signal, bufferSize, ...options } = pipeOpts ?? {};    const outChan = new Channel<void>(      bufferSize ?? target.bufferSize,      options,    );    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;        await fn(res[0]);      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("forEach", fn, err))      .finally(() => outChan.close());    return outChan;  };}
export function filter<T>(  fn: (val: T) => boolean | Promise<boolean>,  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<T>) => {    const { signal, bufferSize, ...options } = pipeOpts ?? {};    const outChan = new Channel<T>(bufferSize ?? target.bufferSize, options);    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;        if (!(await fn(res[0]))) continue;        await outChan.send(res[0], makeAbortCtrl(signal));      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("filter", fn, err))      .finally(() => outChan.close());    return outChan;  };}
export function reduce<T>(  fn: (prev: T, current: T) => T | Promise<T>,  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<T>) => {    const { signal, bufferSize, ...options } = pipeOpts ?? {};    const outChan = new Channel<T>(bufferSize ?? target.bufferSize, options);
    (async () => {      let prev: T;
      const res = await target.get(makeAbortCtrl(signal));      if (!res[1]) return;      prev = res[0];
      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) {          return await outChan.send(prev, makeAbortCtrl(signal));        }        prev = await fn(prev, res[0]);      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("reduce", fn, err))      .finally(() => outChan.close());
    return outChan;  };}
export function groupBy<T, TKey extends (string | symbol)>(  fn: (val: T) => TKey | Promise<TKey>,  pipeOpts?: ChannelPipeOptions,) {  return (target: Receiver<T>) => {    const { signal, bufferSize, debugExtra, ...options } = pipeOpts ?? {};    const out = recordWithDefaults(      {} as Record<TKey, Channel<T>>,      (prop) => {        return new Channel<T>(bufferSize ?? target.bufferSize, {          ...options,          debugExtra: { prop, ...debugExtra },        });      },    );
    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;        const key = await fn(res[0]);        await out[key].send(res[0], makeAbortCtrl(signal));      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("groupBy", fn, err))      .finally(() => {        Object.values<Channel<T>>(out).forEach((ch) => ch.close());      });
    return out;  };}
export type ChannelDuplicateSendMode =  | "WaitForAll"  | "WaitForOne"  | "ContinueImmediately";
export interface ChannelDuplicateOptions extends ChannelPipeOptions {  sendMode: ChannelDuplicateSendMode;}
export function duplicate<T>(n = 2, pipeOpts?: ChannelDuplicateOptions) {  return (target: Receiver<T>) => {    const {      sendMode: sendModesTmp,      signal,      bufferSize,      debugExtra,      ...options    } = pipeOpts ?? {};    const sendMode = sendModesTmp ?? "WaitForAll";    if (!isPositiveSafeInteger(n)) {      throw new TypeError(`${n} is not a safe integer larger than 1`);    }    if (n < 2) throw new RangeError(`${n} is not a safe integer larger than 1`);    const arrOut = [] as Channel<T>[];    for (let i = 0; i < n; i++) {      arrOut[i] = new Channel(bufferSize ?? target.bufferSize, {        ...options,        debugExtra: { i, ...debugExtra },      });    }
    (async () => {      while (true) {        const res = await target.get(makeAbortCtrl(signal));        if (!res[1]) return;
        switch (sendMode) {          case "WaitForAll":            await Promise.all(              arrOut.map((ch) => ch.send(res[0], makeAbortCtrl(signal))),            );            break;
          case "WaitForOne":            await Promise.race(              arrOut.map((ch) => ch.send(res[0], makeAbortCtrl(signal))),            );            break;
          case "ContinueImmediately":            Promise.all(              arrOut.map((ch) => ch.send(res[0], makeAbortCtrl(signal))),            )              .catch((err) => target.error("duplicate", n, err));            break;        }      }    })().catch(ignoreAbortedError)      .catch((err) => target.error("duplicate", n, err))      .finally(() => arrOut.forEach((ch) => ch.close()));
    return arrOut;  };}