reduce.ts

import { ReduceOptions } from './reduceoptions';
import { wrapWithAbort } from './operators/withabort';

export async function reduce<T, R = T>(
  source: AsyncIterable<T>,
  options: ReduceOptions<T, R>
): Promise<R> {
  const { seed, signal, callback } = options;
  const hasSeed = options.hasOwnProperty('seed');
  let i = 0;
  let hasValue = false;
  let acc = seed as T | R;
  for await (const item of wrapWithAbort(source, signal)) {
    if (hasValue || (hasValue = hasSeed)) {
      acc = await callback(<R>acc, item, i++, signal);
    } else {
      acc = item;
      hasValue = true;
      i++;
    }
  }

  if (!(hasSeed || hasValue)) {
    throw new Error('Sequence contains no elements');
  }

  return acc as R;
}

reduceoptions.ts

import { AbortSignal } from '../abortsignal';
import { ScanOptions } from './operators/scanoptions';

export interface ReduceOptions<T, R> extends ScanOptions<T, R> {
  signal?: AbortSignal;
}

scan.ts

import { AbortSignal } from '../../abortsignal';
import { AsyncIterableX } from '../asynciterablex';
import { OperatorAsyncFunction } from '../../interfaces';
import { wrapWithAbort } from './withabort';
import { ScanOptions } from './scanoptions';

export class ScanAsyncIterable<T, R> extends AsyncIterableX<R> {
  private _source: AsyncIterable<T>;
  private _fn: (acc: R, x: T, index: number, signal?: AbortSignal) => R | Promise<R>;
  private _seed?: T | R;
  private _hasSeed: boolean;

  constructor(
    source: AsyncIterable<T>,
    options: ScanOptions<T, R>
  ) {
    super();
    this._source = source;
    this._fn = options.callback;
    this._hasSeed = options.hasOwnProperty('seed');
    this._seed = options.seed;
  }

  async *[Symbol.asyncIterator](signal?: AbortSignal) {
    let i = 0;
    let hasValue = false;
    let acc = this._seed;
    for await (const item of wrapWithAbort(this._source, signal)) {
      if (hasValue || (hasValue = this._hasSeed)) {
        acc = await this._fn(<R>acc, item, i++, signal);
        yield acc;
      } else {
        acc = item;
        hasValue = true;
        i++;
      }
    }
    if (i === 1 && !this._hasSeed) {
      yield acc as R;
    }
  }
}

export function scan<T, R = T>(
  options: ScanOptions<T, R>
): OperatorAsyncFunction<T, R> {
  return function scanOperatorFunction(source: AsyncIterable<T>): AsyncIterableX<R> {
    return new ScanAsyncIterable(source, options);
  };
}

scanoptions.ts

import { AbortSignal } from '../../abortsignal';

export interface ScanOptions<T, R> {
  seed?: R;
  callback: (accumulator: R, current: T, index: number, signal?: AbortSignal) => R | Promise<R>;
}