<commit_before>interface Queue<T> extends AsyncIterableIterator<T> {
  unshift(value: T);
  done(returnValue?: T);
  error(e: any);
  readonly pushingCount: number;
  readonly pullingCount: number;
  readonly isPushing: boolean;
}

export const createQueue = <T>(): Queue<T> => {
  const _pulling: Array<[(r: IteratorResult<T>) => any, (r: any) => any]> = [];
  const _pushing = [];
  let _e: any;
  let _done: boolean;
  let _isPushing: boolean;

  const write = (value: T, done = false) => {
    _isPushing = true;
    return new Promise((resolve, reject) => {
      if (_pulling.length) {
        _pulling.shift()[0]({ value, done });
        _isPushing = false;
        resolve();
      } else {
        _pushing.push(() => {
          resolve();
          _isPushing = _pushing.length === 0;
          return Promise.resolve({ value, done });
        });
      }
    });
  };

  return {
    get isPushing() {
      return _isPushing;
    },
    get pushingCount() {
      return _pushing.length;
    },
    get pullingCount() {
      return _pulling.length;
    },
    [Symbol.asyncIterator]() {
      return this;
    },
    next(value: T): Promise<IteratorResult<T>> {
      if (_e) {
        return Promise.reject(_e);
      }
      if (_pushing.length) {
        return _pushing.shift()();
      }
      if (_done) {
        return Promise.resolve({ done: true });
      }
      return new Promise<IteratorResult<T>>((resolve, reject) => {
        _pulling.push([resolve, reject]);
      });
    },
    unshift(value: T) {
      return write(value);
    },
    done(returnValue) {
      if (_done) {
        return Promise.resolve();
      }
      _done = true;
      return write(returnValue, true);
    },
    error(e) {
      _e = e;
      if (_pulling.length) {
        _pulling.shift()[1](e);
      }
    }
  }
};

export interface DuplexStream<TInput, UOutput> extends AsyncIterableIterator<UOutput> {
  next(input?: TInput): Promise<IteratorResult<UOutput>>;
}

export const createDuplexStream = <TInput, TOutput>(): DuplexStream<TInput, TOutput> & { input: Queue<TInput>, output: Queue<TOutput> } => {
  const input  = createQueue<TInput>();
  const output = createQueue<TOutput>();

  return {
    [Symbol.asyncIterator]: () => this,
    input,
    output,
    next(value: TInput) {
      input.unshift(value);
      return output.next();
    }
  };
}

// pipe(async function*() => yield 1, duplex)

export function readAll<T>(stream: DuplexStream<any, T>): Promise<T[]>;
export function readAll<T>(stream: AsyncIterableIterator<T>): Promise<T[]>;

export function readAll<T>(source: any): Promise<T[]> {
  return new Promise((resolve, reject) => {
    const result = [];
    pump(source, (chunk) => result.push(chunk)).then(resolve.bind(this, result)).catch(reject);
  });
}

export function pipe(...pipeline: any[]): AsyncIterableIterator<any>;
export function pipe(...pipeline: IterableIterator<any>[]): AsyncIterableIterator<any>;
export function pipe(...pipeline: AsyncIterableIterator<any>[]): AsyncIterableIterator<any>;

export function pipe<TInput, TOutput>(...pipeline: any[]) {

  let _done = false;

  const targets = pipeline.map(wrapAsyncIterableIterator);

  const call = (methodName: string, value: TInput) => {
    return new Promise((resolve, reject) => {
      const remaining = targets.concat();
      const next = ({ value, done }) => {

        if (!_done) {
          _done = done;
        }

        // if one piped item finishes, then we need to finish
        if (!remaining.length || _done) return resolve({value, done});
        const fn = remaining.shift()[methodName];
        return fn ? fn(value).then(next, reject) : next(value);
      }
      next({ value, done: false });
    });
  };

  return {
    [Symbol.asyncIterator]: () => this,
    next: call.bind(this, "next"),
    return: call.bind(this, "return"),
    throw: call.bind(this, "throw")
  };
}

export function through<TInput, TOutput>(fn: (input: TInput) => TOutput, keepOpen?: boolean): AsyncIterableIterator<TOutput>;
export function through<TInput, TOutput>(fn: (input: TInput) => IterableIterator<TOutput>, keepOpen?: boolean): AsyncIterableIterator<TOutput>;
export function through<TInput, TOutput>(fn: (input: TInput) => AsyncIterableIterator<TOutput>, keepOpen?: boolean): AsyncIterableIterator<TOutput>;

// TODO - possibly endOnNoInput
export function through<TInput>(fn: (input: TInput) => any, keepOpen: boolean = false) {
  let _running: boolean;
  let _pumping: boolean;
  const outputQueue = createQueue();
  const inputQueue  = createQueue<TInput>();

  const run = () => {
    if (_running) {
      return;
    }
    _running = true;

    const nextInput = () => {
      inputQueue.next().then(({value}) => {
        _pumping = true;
        return pump(wrapAsyncIterableIterator(fn(value)), (value) => {
          return outputQueue.unshift(value);
        }).then(() => {
          _pumping = false;
          nextInput();
        });
      });
    };

    nextInput();
  }


  function next(value?: TInput) {
    if (value != null) {
      run();
      inputQueue.unshift(value);
    } else if (!keepOpen) {
      Promise.resolve().then(() => {
        if (!_pumping) {
          outputQueue.done();
        }
      });
    }

    return outputQueue.next();
  }

  return {
    [Symbol.asyncIterator]: () => this,
    next: next
  };
}

/**<commit_msg>add defer promise util<commit_after>interface Queue<T> extends AsyncIterableIterator<T> {
  unshift(value: T);
  done(returnValue?: T);
  error(e: any);
  readonly pushingCount: number;
  readonly pullingCount: number;
  readonly isPushing: boolean;
}

export const createQueue = <T>(): Queue<T> => {
  const _pulling: Array<[(r: IteratorResult<T>) => any, (r: any) => any]> = [];
  const _pushing = [];
  let _e: any;
  let _done: boolean;
  let _isPushing: boolean;

  const write = (value: T, done = false) => {
    _isPushing = true;
    return new Promise((resolve, reject) => {
      if (_pulling.length) {
        _pulling.shift()[0]({ value, done });
        _isPushing = false;
        resolve();
      } else {
        _pushing.push(() => {
          resolve();
          _isPushing = _pushing.length === 0;
          return Promise.resolve({ value, done });
        });
      }
    });
  };

  return {
    get isPushing() {
      return _isPushing;
    },
    get pushingCount() {
      return _pushing.length;
    },
    get pullingCount() {
      return _pulling.length;
    },
    [Symbol.asyncIterator]() {
      return this;
    },
    next(value: T): Promise<IteratorResult<T>> {
      if (_e) {
        return Promise.reject(_e);
      }
      if (_pushing.length) {
        return _pushing.shift()();
      }
      if (_done) {
        return Promise.resolve({ done: true });
      }
      return new Promise<IteratorResult<T>>((resolve, reject) => {
        _pulling.push([resolve, reject]);
      });
    },
    unshift(value: T) {
      return write(value);
    },
    done(returnValue) {
      if (_done) {
        return Promise.resolve();
      }
      _done = true;
      return write(returnValue, true);
    },
    error(e) {
      _e = e;
      if (_pulling.length) {
        _pulling.shift()[1](e);
      }
    }
  }
};

export const createDeferredPromise = <T>() => {
  let resolve;
  let reject;
  let promise: Promise<T>;

  promise = new Promise((resolve, reject) => {
    resolve = resolve;
    reject = reject;
  });
  
  return {
    promise,
    resolve,
    reject,
  }
}

export interface DuplexStream<TInput, UOutput> extends AsyncIterableIterator<UOutput> {
  next(input?: TInput): Promise<IteratorResult<UOutput>>;
}

export const createDuplexStream = <TInput, TOutput>(): DuplexStream<TInput, TOutput> & { input: Queue<TInput>, output: Queue<TOutput> } => {
  const input  = createQueue<TInput>();
  const output = createQueue<TOutput>();

  return {
    [Symbol.asyncIterator]: () => this,
    input,
    output,
    next(value: TInput) {
      input.unshift(value);
      return output.next();
    }
  };
}

// pipe(async function*() => yield 1, duplex)

export function readAll<T>(stream: DuplexStream<any, T>): Promise<T[]>;
export function readAll<T>(stream: AsyncIterableIterator<T>): Promise<T[]>;

export function readAll<T>(source: any): Promise<T[]> {
  return new Promise((resolve, reject) => {
    const result = [];
    pump(source, (chunk) => result.push(chunk)).then(resolve.bind(this, result)).catch(reject);
  });
}

export function pipe(...pipeline: any[]): AsyncIterableIterator<any>;
export function pipe(...pipeline: IterableIterator<any>[]): AsyncIterableIterator<any>;
export function pipe(...pipeline: AsyncIterableIterator<any>[]): AsyncIterableIterator<any>;

export function pipe<TInput, TOutput>(...pipeline: any[]) {

  let _done = false;

  const targets = pipeline.map(wrapAsyncIterableIterator);

  const call = (methodName: string, value: TInput) => {
    return new Promise((resolve, reject) => {
      const remaining = targets.concat();
      const next = ({ value, done }) => {

        if (!_done) {
          _done = done;
        }

        // if one piped item finishes, then we need to finish
        if (!remaining.length || _done) return resolve({value, done});
        const fn = remaining.shift()[methodName];
        return fn ? fn(value).then(next, reject) : next(value);
      }
      next({ value, done: false });
    });
  };

  return {
    [Symbol.asyncIterator]: () => this,
    next: call.bind(this, "next"),
    return: call.bind(this, "return"),
    throw: call.bind(this, "throw")
  };
}

export function through<TInput, TOutput>(fn: (input: TInput) => TOutput, keepOpen?: boolean): AsyncIterableIterator<TOutput>;
export function through<TInput, TOutput>(fn: (input: TInput) => IterableIterator<TOutput>, keepOpen?: boolean): AsyncIterableIterator<TOutput>;
export function through<TInput, TOutput>(fn: (input: TInput) => AsyncIterableIterator<TOutput>, keepOpen?: boolean): AsyncIterableIterator<TOutput>;

// TODO - possibly endOnNoInput
export function through<TInput>(fn: (input: TInput) => any, keepOpen: boolean = false) {
  let _running: boolean;
  let _pumping: boolean;
  const outputQueue = createQueue();
  const inputQueue  = createQueue<TInput>();

  const run = () => {
    if (_running) {
      return;
    }
    _running = true;

    const nextInput = () => {
      inputQueue.next().then(({value}) => {
        _pumping = true;
        return pump(wrapAsyncIterableIterator(fn(value)), (value) => {
          return outputQueue.unshift(value);
        }).then(() => {
          _pumping = false;
          nextInput();
        });
      });
    };

    nextInput();
  }

  function next(value?: TInput) {
    if (value != null) {
      run();
      inputQueue.unshift(value);
    } else if (!keepOpen) {
      Promise.resolve().then(() => {
        if (!_pumping) {
          outputQueue.done();
        }
      });
    }

    return outputQueue.next();
  }

  return {
    [Symbol.asyncIterator]: () => this,
    next: next
  };
}

/**
{'fraction_non_alphanumeric': 0.15838087476229284, 'fraction_numerical': 0.0007244408222403332, 'mean_word_length': 2.88736360436466, 'pattern_counts': {'":': 0, '<': 93, '<?xml version=': 0, '>': 149, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7039364', 'n_tokens_mistral': 3481, 'n_tokens_neox': 3435, 'n_words': 794}