Discussion on: Limit concurrent asynchronous calls

View post

I implemented this whole thing with a working code, let me know if I am doing something incorrectly.

Note: I am using setTimeout for async calls so promise failure will not happen.

const ASYNC_LIMIT = 2;

function scheduler(cb, id, delay) {
    return new Promise((resolve, reject) => {
        setTimeout(() => {
            cb(id);
            resolve();
        }, delay);
    });
}

const asyncLimit = (fn, n) => {
    let promiseArray = [];

    return async function(...args) {
        if (promiseArray.length >= n) {
            await Promise.race(promiseArray);
        }

        let p = fn.call(this, ...args);
        promiseArray.push(p);

        p.then(() => {
            promiseArray = promiseArray.filter(pending => p !== pending);
        });

        return p;
    };
};

let cb = id => {
    console.log(id + " task completed", Date.now() % 10000);
};

let modifiedScheduler = asyncLimit(scheduler, ASYNC_LIMIT);

modifiedScheduler(cb, 1, 5000);
modifiedScheduler(cb, 2, 2000);
modifiedScheduler(cb, 3, 1500);
modifiedScheduler(cb, 4, 3000);
modifiedScheduler(cb, 5, 4000);
modifiedScheduler(cb, 6, 1000);
modifiedScheduler(cb, 7, 2500);

tconrado • May 20 '20

using the corrections from @kusha and the data and functino of your code it did work as expected!
hurray!

exceptional implementation to deal with REST API

Shubham Mishra • May 21 '20

Can you post your working snippet here?

tconrado • Jun 13 '20 • Edited

hey I'm terrible with markdown, but the example bellow is in a lib, basically it does limit the number of active connections to a http API rest service to the a defined number (8 in this case), so a the http request is resolved, it start another connection keeping always 8 active connections to the API server; it will completely hide the connection/handshake delay while guarantee no 429 error (too many requests); from my experience fastest safe approach as you can know the maximum number of calls per second of the API service


// this is the asyncLimit adjusted 
const asyncLimit = (fn, n) => {
  const pendingPromises = new Set();
  return async function(...args) {
    while (pendingPromises.size >= n) {
      await Promise.race(pendingPromises);
    }
    const p = fn.apply(this, args);
    const r = p.catch(() => {});
    pendingPromises.add(r);
    await r;
    pendingPromises.delete(r);
    return p;
  };
};

// native node.js https module to connect to shopify servers
const https = require('https')
exports.httpRequest = function(method, path, body = null) {
  const reqOpt = { 
    method: method,
    path: '/admin' + path,
    hostname: 'xxxxxxxxxxxxxxxxxxxx.myshopify.com', 
    headers: {
      "Content-Type": "application/json",
      "X-Shopify-Access-Token": "xxxxxxxxxxxxxxxxxxxx",
      'Cookie': '',
      "Cache-Control": "no-cache"
    }
  }
  if (body) reqOpt.headers['Content-Length'] = Buffer.byteLength(body);
  return new Promise((resolve, reject) => {

      const clientRequest = https.request(reqOpt, incomingMessage => {
          let response = {
              statusCode: incomingMessage.statusCode,
              headers: incomingMessage.headers,
              body: []
          };
          let chunks = ""
          incomingMessage.on('data', chunk => { chunks += chunk; });
          incomingMessage.on('end', () => {
              if (chunks) {
                  try {
                      response.body = JSON.parse(chunks);
                  } catch (error) {
                      reject(error)
                  }
              }
              resolve(response);
          });
      });
      clientRequest.on('error', error => { reject(error); });
      if (body) { clientRequest.write(body)  }  
      clientRequest.end();

  });
}


// the number 8 bellow can be changed to match the REST API service limits
// assume that this amount will call at once and will be replaced dynamically, hence
// if the service limit it 20 calls per second, be aware that 8 calls will hit the service at once
// using 40% of the maximum (avoid going higher)

exports.ratedhttpRequest = asyncLimit(exports.httpRequest, 8);

tconrado • May 20 '20

hey, tried your code, did not work...

so, assigned the same delay for all of those...like 2 seconds... I did expect to see it process 2, schedule + 2; so, the result would be observable every 2 seconds and everytime 2 task