batch throttle

index.ts

type Deferred<T> = {
  resolve: (value: T) => void;
  reject: (reason?: any) => void;
  promise: Promise<T>;
}

// Call function when timeout is reached
// by splitin array of arguments into batches
// and calling function with each batch
// Allow setup maxParallel to limit number of parallel calls
export function batchThrottle<Parameters extends any[], Return>(fn: (args: Parameters[]) => (Return[] | Promise<Return[]>), {
  timeout = 10,
  size = 1000,
  maxParallel = 2
} = {}) {
  let timerId: NodeJS.Timeout | null = null;
  const deferredQueue: Deferred<Return[] | Promise<Return[]>>[] = [];
  const args: Parameters[] = [];

  return (...callArgs: Parameters) => {
    const callIndex = args.length % size;
    args.push(callArgs);

    if (timerId === null) {
      timerId = setTimeout(flush, timeout);
    }

    if (callIndex === 0) {
      deferredQueue.push(createDeferred());
    }

    const promise = deferredQueue[deferredQueue.length - 1].promise;
    return promise.then(results => results[callIndex]);
  }

  async function flush() {
    clearTimeout(timerId!);
    timerId = null;
    while (deferredQueue.length > 0) {
      await Promise.all(
        deferredQueue.splice(0, maxParallel).map((deferred) => {
          try {
            const batchArgs = args.splice(0, size);
            deferred.resolve(fn(batchArgs));
          } catch (error) {
            deferred.reject(error);
          }
          return deferred.promise;
        })
      );
    }
  }

  function createDeferred<T>(): Deferred<T> {
    let resolve: ((value: T) => void) | null = null;
    let reject: ((reason?: any) => void) | null = null;
    const promise = new Promise<T>((_resolve, _reject) => {
      resolve = _resolve;
      reject = _reject;
    });

    return {
      resolve: resolve!,
      reject: reject!,
      promise
    };
  }
}

// Call function imidiately when size is reached
// or wiat for timeout to be reached
export function batchThrottle<Parameters extends any[], Return>(fn: (args: Parameters[]) => (Return[] | Promise<Return[]>), {
  timeout = 10,
  size = 1000
} = {}) {
  let resolve: ((value: Return[] | Promise<Return[]>) => void) | null = null;
  let reject: ((reason?: any) => void) | null = null;
  let promise: Promise<Return[]> | null = null;
  let timerId: NodeJS.Timeout | null = null;
  const args: Parameters[] = [];
  const flush = () => {
    clearTimeout(timerId!);
    promise = null;
    timerId = null;

    if (resolve === null || reject === null) {
      return;
    }

    const _reject = reject;
    const _resolve = resolve;
    resolve = null;
    reject = null;
    try {
      const argsCopy = args.slice();
      args.length = 0;
      _resolve(fn(argsCopy));
    } catch (error) {
      _reject(error);
    }
  };

  return (...callArgs: Parameters) => {
    const callIndex = args.length;
    args.push(callArgs);

    if (promise === null) {
      promise = new Promise<Return[]>((_resolve, _reject) => {
        resolve = _resolve;
        reject = _reject;
      });
      timerId = setTimeout(flush, timeout);
    }

    const result = promise.then(results => {
      return results[callIndex];
    });

    if (args.length >= size) {
      flush();
    }

    return result;
  }
}