invokeCallsWithLimit.js

// invokeCallsWithLimit.js
// ------------------------------------------------------------------
//

/* jshint esversion:6, node:false, strict:implied */
/* global properties, httpClient, context, Request */

// Specify the limit of concurrent outbound requests 
const ASYNC_LIMIT = 3;

// ===========
// Provide these two functions
function indexedOnComplete(id, response, error) {
  context.setVariable('jsresponse.'+ zeroExtend(id) +'.recd', String(Date.now() - start));
    if (response) {
      context.setVariable('jsresponse.'+ zeroExtend(id) +'.status', response.status);
    }
    else {
      context.setVariable('jsresponse.'+ zeroExtend(id) +'.error', 'Whoops: ' + error);
    }
}

function invokeOne(id, requestContext, cb) {
  // context can hold anything; in this simple example it holds only a delay
  var delay = requestContext.delay;
  var url = properties.target + '/delay?stddev=0.2&mean=' + delay ;
  var headers = { 'Request-ID' : id }; // for diagnostics only, your headers can be anything
  var req = new Request(url, 'GET', headers);
  httpClient.send(req, cb);
}
// ==============



var start = Date.now();

function zeroExtend(n) {
  return ('00' + n).slice(-2);
}

function asyncLimit(fn, cb, n) {
  var requestList = [];

  function inState(state){
    return requestList.filter(function(item) {
            return item.state == state;
          });
  }

  return function(id) {
    var args = Array.prototype.slice.call(arguments);
    var wrapperCb = function (response, error) {
          requestList[id].state = 'finished';
          cb(id, response, error);

          // if any requests are queued (should be at most one), handle the first one
          var queuedRequests = inState('queued');
          if (queuedRequests.length > 0) {
            queuedRequests[0].state = 'pending';
            fn.apply(this, queuedRequests[0].args);
          }
        };

    // inject a wrapper callback
    args.push(wrapperCb);

    // send or queue the current request
    var pendingRequests = inState('pending');
    if (pendingRequests.length < n) {
      // send immediately
      requestList.push({id:id, state: 'pending'});
      fn.apply(this, args);
    }
    else {
      // queue for send later
      requestList.push({id:id, state:'queued', args:args});
    }
  };
}

const limitedTask = asyncLimit(invokeOne, indexedOnComplete, ASYNC_LIMIT);

const reducer =
  function(acc, cur, ix) {
    return limitedTask(ix, cur); 
  };

// provide the array. One request per element in the array
const a = [
        { delay: 1.2},
        { delay: 0.5},
        { delay: 0.75},
        { delay: 0.9},
        { delay: 0.75},
        { delay: 1.3},
        { delay: 0.3},
        { delay: 0.8},
        { delay: 1.2},
        { delay: 0.9},
        { delay: 0.75},
        { delay: 0.3},
        { delay: 1 }
      ];

a.reduce(reducer, 0);