Guard for gracefully handling errors thrown during async iterable iteration

guard.test.ts

import { collect } from 'streaming-iterables';
import { guard, mapGuard } from './guard';

describe('mapGuard()', () => {
  it('Should handle errors thrown while mapping.', async () => {
    const err = new Error('An error');
    const mapper = (throwError: boolean) => {
      if (throwError) {
        throw err;
      }
      return 1;
    };
    const onError = jest.fn();

    const result = await collect(
      mapGuard(mapper, onError, [false, false, true, false]),
    );
    expect(result).toEqual([1, 1, 1]);
    expect(onError).toHaveBeenCalledTimes(1);
    expect(onError).toHaveBeenCalledWith(err, true);
  });
});

describe('guard()', () => {
  it('Should prevent errors thrown during iteration from being thrown', async () => {
    const err = new Error('An error');
    const producer = async function* () {
      yield 1;
      yield 1;
      throw err;
      yield 1;
    };
    const onError = jest.fn();
    const i = guard(onError, producer());
    const result = await collect(guard(onError, i));
    expect(result).toEqual([1, 1]);
    expect(onError).toHaveBeenCalledTimes(1);
    expect(onError).toHaveBeenCalledWith(err);
  });
});

guard.ts

import { AnyIterable } from 'streaming-iterables';

type GuardErrorHandler = (e: unknown) => void | Promise<void>;
async function* _guard<T>(
  onError: GuardErrorHandler,
  iterable: AnyIterable<T>,
): AsyncIterableIterator<T> {
  try {
    yield* iterable;
  } catch (e) {
    await onError(e);
  }
}

/**
 * This function is useful if you have an iterable where you expect an error from one of the
 * iterations and would like to handle it more gracefully than a simple thrown error.
 *
 * It's particularly designed for sewing many iterables together, where you want to keep going
 * even if a particular iterator fails during iteration.
 *
 * @example
 * function parseFile(filename) {
 *     const input = readFileToStream();
 *     const parsed = input.pipe(parse());
 *     const handleError = (e) => console.error(`An error occurred during parsing of ${filename}: ${e}`);
 *     // Parse errors will no longer cause an exception - they'll trigger a log entry.
 *     return guard(handleError, parsed);
 * }
 *
 * @param onError
 */
export function guard<T>(
  onError: GuardErrorHandler,
): (iterable: AnyIterable<T>) => AsyncIterableIterator<T>;
export function guard<T>(
  onError: GuardErrorHandler,
  iterable: AnyIterable<T>,
): AsyncIterableIterator<T>;
export function guard<T>(
  onError: GuardErrorHandler,
  iterable?: AnyIterable<T>,
):
  | ((iterable: AnyIterable<T>) => AsyncIterableIterator<T>)
  | AsyncIterableIterator<T> {
  if (iterable === undefined) {
    return (curriedIterable: AnyIterable<T>) =>
      _guard(onError, curriedIterable);
  }
  return _guard(onError, iterable);
}

type MapGuardErrorHandler<T> = (e: unknown, item: T) => void | Promise<void>;
type MapFn<T, B> = (data: T) => B | Promise<B>;
async function* _mapGuard<T, B>(
  func: MapFn<T, B>,
  onError: MapGuardErrorHandler<T>,
  iterable: AnyIterable<T>,
): AsyncIterableIterator<B> {
  for await (const item of iterable) {
    try {
      yield await func(item);
    } catch (e) {
      await onError(e, item);
      // handle errors here.
    }
  }
}

/**
 * This function is useful for allowing a map() function to be called many times, where you know
 * some of the invocations might throw an error you'd like to catch.
 *
 * Note: If the map function actually RETURNS an async iterable, errors thrown during iteration
 * over that object will not be caught here.
 *
 * @param func
 * @param onError
 */

export function mapGuard<T, B>(
  func: MapFn<T, B>,
  onError: MapGuardErrorHandler<T>,
): (iterable: AnyIterable<T>) => AsyncIterableIterator<B>;
export function mapGuard<T, B>(
  func: MapFn<T, B>,
  onError: MapGuardErrorHandler<T>,
  iterable: AnyIterable<T>,
): AsyncIterableIterator<B>;
export function mapGuard<T, B>(
  func: MapFn<T, B>,
  onError: MapGuardErrorHandler<T>,
  iterable?: AnyIterable<T>,
):
  | ((iterable: AnyIterable<T>) => AsyncIterableIterator<B>)
  | AsyncIterableIterator<B> {
  if (iterable === undefined) {
    return (curriedIterable: AnyIterable<T>) =>
      _mapGuard(func, onError, curriedIterable);
  }
  return _mapGuard(func, onError, iterable);
}