TypeScript: Queueing promises

queuePromise.ts

export function bufferPromise(promiseFactoryFn: any, maxActive = 5) {
  const pendingPromises: Promise<any>[] = [];
  const waitingPromises: Array<() => Promise<any>> = [];

  function promiseCompleted(promise: Promise<any>) {
    const completedPromiseIndex = pendingPromises.findIndex(
      pendingPromise => pendingPromise === promise
    );

    pendingPromises.splice(completedPromiseIndex, 1);

    if (waitingPromises.length) {
      const nextPromise = waitingPromises.splice(0, 1)[0];
      pendingPromises.push(nextPromise());
    }
  }

  function executePromise(
    payload: any,
    resolve: (result?: any) => any,
    reject: (error?: any) => any
  ) {
    const promise = promiseFactoryFn(payload);

    promise
      .then((result: any) => {
        promiseCompleted(promise);
        resolve(result);
      })
      .catch((error: any) => {
        promiseCompleted(promise);
        reject(error);
      });

    pendingPromises.push(promise);

    return promise;
  }

  function queuePromise(
    payload: any,
    resolve: (result?: any) => any,
    reject: (error?: any) => any
  ) {
    return () => executePromise(payload, resolve, reject);
  }

  function schedulePromise(
    payload: any,
    resolve: (result?: any) => any,
    reject: (error?: any) => any
  ) {
    if (pendingPromises.length === maxActive)
      waitingPromises.push(queuePromise(payload, resolve, reject));
    else executePromise(payload, resolve, reject);
  }

  return (payload?: any) => {
    return new Promise((resolve, reject) => {
      schedulePromise(payload, resolve, reject);
    });
  };
}

test.ts

import {expect} from "chai";
import {bufferPromise} from "./solution";

interface Resolve {
  id: number,
  resolve: (v: any) => any,
  reject: (v: any) => any,
}
interface CFG {
  calls: number,
  resolves: Resolve[],
  rejects: Resolve[],
}

function promiseFn(cfg: CFG) {
  expect(cfg).to.be.an('object');
  cfg.calls++;
  return new Promise((resolve, reject) => {
    cfg.resolves.push({
      id: cfg.calls,
      resolve: (v: any) => setTimeout(() => resolve(v)),
      reject: (v: any) => setTimeout(() => reject(v)),
    });
  });
}
describe('bufferPromise', function() {
  function getCFG(): CFG {
    return {calls: 0, resolves: [], rejects: []};
  };
  
  it('should work like a normal async function', function() {
    let cfg = getCFG();
    let bp = bufferPromise(promiseFn);
    let p = bp(cfg);
    cfg.resolves[0].resolve(true);
    return p.then((resp: any) => {
      expect(resp).to.be.true;
    });
  });
  
  it('should handle rejections', function() {
    let cfg = getCFG();
    let bp = bufferPromise(promiseFn);
    let p = bp(cfg);
    cfg.resolves[0].reject('rejected');
    return p.then(() => {
      throw new Error('should not resolve');
    }).catch((err: any) => {
      expect(err).to.eql('rejected');
    });
  });
  
  describe('should buffer additional calls with rejections (1 at a time), resolved [1, 2, 3]', function() {
    let cfg = getCFG();
    let bp = bufferPromise(promiseFn, 1);
    let p1 = bp(cfg);
    let p2 = bp(cfg);
    let p3 = bp(cfg);
    let counter = 0;
    
    it('should initiate the promises correctly', function() {
      expect(cfg.calls).to.eql(1, 'Promise factory should have only been called 1 time');
    });
    
    it('should handle Promise #1', function() {
      cfg.resolves[0].resolve(1);
      return p1.then((resp: any) => {
        expect(resp).to.eql(1, 'correct result passed back');
        counter++;
        expect(counter).to.eql(1, 'results happened in order');
      })
      .then(() => {
        expect(cfg.calls).to.eql(2, 'promise factory was only called 2 times');
      });
    });
    
    it('should handle Promise #2 (rejected)', function() {
      cfg.resolves[1].reject(2);
      return p2.then(() => {
        throw new Error('should not resolve');
      }).catch((err: any) => {
        console.log('cfg', cfg);
        expect(err).to.eql(2, 'correct rejection passed back');
        counter++;
        expect(counter).to.eql(2, 'results happened in order');
        cfg.resolves[2].resolve(3);
      })
      .then(() => {
        expect(cfg.calls).to.eql(3, 'promise factory was called 3 times');
      });
    });
    
    it('should handle Promise #3', function() {
      return p3.then((resp: any) => {
        expect(resp).to.eql(3, 'correct result passed back');
        counter++;
        expect(counter).to.eql(3, 'results happened in order');
      })
      .then(() => {
        expect(cfg.calls).to.eql(3, 'promise factory was not called again');
      });
    });
  });
});

TypeScript: Queueing promises.md

What's this?

This was my answer to an exam that I went through online today. I think it might also be useful in the future so I'm putting it here as future reference when I need this functionality.

Buffering Promises

In this challenge, you'll be asked to wrap an unknown asynchronous function, returning a new function with a modified asynchronous behavior. The returned function will prevent making too many calls to the asynchronous function at once by buffering extra requests until an earlier one has completed.

A common use case would be to prevent overrunning an API endpoint, so you make no more than 4 requests simultaneously.

Your Task

Modify the bufferPromise function to enable calling up to a maxActive times to our external deferred process, buffering any more requests until we have less than maxActive requests running.

Example usage

function loadRemote(url) {
    return Promise(...);
}

// your function is bufferPromise
let bufferedLoadRemote = bufferPromise(loadRemote, 4);

// many urls are requested, but only 4 are active at a time
let urls = [...].map(bufferedLoadRemote);

// we could even make an additional request, without worrying
// about the buffering process
urls.push(bufferedLoadRemote('http://someOtherUrl'));

// As an example, the end user might wait on every promise.
// (Note: even if an earlier promise rejects, every following
// request should go through)
return Promise.all(urls);

Details

• The external process is initiated via promiseFactoryFn, which returns a Promise.
• If there are less than maxActive active processes, start the next request immediately.
• If there are maxActive or more processes already running, then postpone the request until at least one of those processes has completed.

Hint: Make sure to start the next process without waiting on the previous promise handler. In general, you want to start the next asynchronous request first, then resolve or reject the promise. This ensures we aren't depending on the promise handler to trigger the next result.

• Each request must be made in the order requested, but the resolutions or rejections may happen in any order.

For example, you might make requests A, B, C, and D, with a maxActive of 2. In this case, A and B will be initiated immediately, and C and D will be buffered. If B then resolves (or rejects), C is started. Then only after A or C resolve or reject do we start D.