lib/requests.js
function sendAll(requests, fnName, concurrency, callback) {
    if (typeof concurrency === 'function') {
      callback = concurrency;
      concurrency = 1;
    }

    var q = queue(concurrency);

    requests.forEach(function(req) {
      q.defer(function(next) {
        if (!req) return next();
        req.on('complete', function(response) {
          next(null, response);
        }).send();
      });
    });

    q.awaitAll(function(err, responses) {
      if (err) return callback(err);

      var errors = [];
      var data = [];
      var unprocessed = [];

      responses.forEach(function(response) {
        if (!response) response = { error: null, data: null };

        errors.push(response.error);
        data.push(response.data);

        if (!response.data) return unprocessed.push(null);

        var newParams = {
          RequestItems: response.data.UnprocessedItems || response.data.UnprocessedKeys
        };

        if (newParams.RequestItems && !Object.keys(newParams.RequestItems).length)
          return unprocessed.push(null);

        unprocessed.push(newParams.RequestItems ? newParams : null);
      });

      unprocessed = unprocessed.map(function(params) {
        if (params) return client[fnName].bind(client)(params);
        else return null;
      });

      var unprocessedCount = unprocessed.filter(function(req) { return !!req; }).length;
      if (unprocessedCount) unprocessed.sendAll = sendAll.bind(null, unprocessed, fnName);

      var errorCount = errors.filter(function(err) { return !!err; }).length;

      callback(
        errorCount ? errors : null,
        data,
        unprocessedCount ? unprocessed : null
      );
    });
  }
function sendAll(requests, fnName, concurrency, callback) {
    if (typeof concurrency === 'function') {
      callback = concurrency;
      concurrency = 1;
    }

    var q = queue(concurrency);

    requests.forEach(function(req) {
      q.defer(function(next) {
        if (!req) return next();
        req.on('complete', function(response) {
          next(null, response);
        }).send();
      });
    });

    q.awaitAll(function(err, responses) {
      if (err) return callback(err);

      var errors = [];
      var data = [];
      var unprocessed = [];

      responses.forEach(function(response) {
        if (!response) response = { error: null, data: null };

        errors.push(response.error);
        data.push(response.data);

        if (!response.data) return unprocessed.push(null);

        var newParams = {
          RequestItems: response.data.UnprocessedItems || response.data.UnprocessedKeys
        };

        if (newParams.RequestItems && !Object.keys(newParams.RequestItems).length)
          return unprocessed.push(null);

        unprocessed.push(newParams.RequestItems ? newParams : null);
      });

      unprocessed = unprocessed.map(function(params) {
        if (params) return client[fnName].bind(client)(params);
        else return null;
      });

      var unprocessedCount = unprocessed.filter(function(req) { return !!req; }).length;
      if (unprocessedCount) unprocessed.sendAll = sendAll.bind(null, unprocessed, fnName);

      var errorCount = errors.filter(function(err) { return !!err; }).length;

      callback(
        errorCount ? errors : null,
        data,
        unprocessedCount ? unprocessed : null
      );
    });
  }
function sendAll ( requests , fnName , concurrency , callback ) { if ( typeof concurrency === 'function' ) { callback = concurrency ; concurrency = 1 ; } var q = queue ( concurrency ) ; requests . forEach ( function ( req ) { q . defer ( function ( next ) { if ( ! req ) return next ( ) ; req . on ( 'complete' , function ( response ) { next ( null , response ) ; } ) . send ( ) ; } ) ; } ) ; q . awaitAll ( function ( err , responses ) { if ( err ) return callback ( err ) ; var errors = [ ] ; var data = [ ] ; var unprocessed = [ ] ; responses . forEach ( function ( response ) { if ( ! response ) response = { error : null , data : null } ; errors . push ( response . error ) ; data . push ( response . data ) ; if ( ! response . data ) return unprocessed . push ( null ) ; var newParams = { RequestItems : response . data . UnprocessedItems || response . data . UnprocessedKeys } ; if ( newParams . RequestItems && ! Object . keys ( newParams . RequestItems ) . length ) return unprocessed . push ( null ) ; unprocessed . push ( newParams . RequestItems ? newParams : null ) ; } ) ; unprocessed = unprocessed . map ( function ( params ) { if ( params ) return client [ fnName ] . bind ( client ) ( params ) ; else return null ; } ) ; var unprocessedCount = unprocessed . filter ( function ( req ) { return ! ! req ; } ) . length ; if ( unprocessedCount ) unprocessed . sendAll = sendAll . bind ( null , unprocessed , fnName ) ; var errorCount = errors . filter ( function ( err ) { return ! ! err ; } ) . length ; callback ( errorCount ? errors : null , data , unprocessedCount ? unprocessed : null ) ; } ) ; }
Given a set of requests, this function sends them all at specified concurrency.
Generally, this function is not called directly, but is bound to an array of
requests with the first two parameters wired to specific values.

@private
@param {RequestSet} requests - an array of AWS.Request objects
@param {string} fnName - the name of the aws-sdk client function that will
be used to construct new requests should an unprocessed items be detected
@param {number} concurrency - the concurrency with which batch requests
will be made
@param {function} callback - a function to handle the response
Given a set of requests this function sends them all at specified concurrency . Generally this function is not called directly but is bound to an array of requests with the first two parameters wired to specific values .