Implements throttling

index.js

@@ -1,8 +1,9 @@
 'use strict';
 
-var Command = require('./lib/Command.js'),
-    request = require('./lib/Request.js'),
-    libxml  = require('libxmljs-dom'),
+var Command     = require('./lib/Command.js'),
+    request     = require('./lib/Request.js'),
+    libxml      = require('libxmljs-dom'),
+    RateLimiter = require('limiter').RateLimiter,
     instanceId      = 0,
     memoryUsage     = 0,
     cachedSelectors = {},
@@ -57,9 +58,10 @@ function Osmosis(url, params) {
         return Osmosis.get(url, params);
     }
 
-    this.queue   = [];
-    this.command = new Command(this);
-    this.id      = ++instanceId;
+    this.queue    = [];
+    this.command  = new Command(this);
+    this.id       = ++instanceId;
+    this.throttle = new RateLimiter(999, 1, true);
 }
 
 /**
@@ -221,54 +223,56 @@ Osmosis.prototype.request = function (url, opts, callback, tries) {
     this.requests++;
     this.stack.requests++;
     this.stack.push();
-
-    request(url.method,
-            url,
-            url.params,
-            opts,
-            tries,
-            function (err, res, data) {
-                var proxies = opts.proxies;
-
-                self.stack.requests--;
-
-                if ((res === undefined || res.statusCode !== 404) &&
-                    proxies !== undefined) {
-                    self.command.error('proxy ' + (proxies.index + 1) +
-                                        '/' + proxies.length +
-                                        ' failed (' + opts.proxy + ')');
-
-                    // remove the failing proxy
-                    if (proxies.length > 1) {
-                        opts.proxies.splice(proxies.index, 1);
-                        opts.proxy = proxies[proxies.index];
+
+    this.throttle.removeTokens(1, function(err, remainingRequests) {
+        request(url.method,
+                url,
+                url.params,
+                opts,
+                tries,
+                function (err, res, data) {
+                    var proxies = opts.proxies;
+
+                    self.stack.requests--;
+
+                    if ((res === undefined || res.statusCode !== 404) &&
+                        proxies !== undefined) {
+                        self.command.error('proxy ' + (proxies.index + 1) +
+                                            '/' + proxies.length +
+                                            ' failed (' + opts.proxy + ')');
+
+                        // remove the failing proxy
+                        if (proxies.length > 1) {
+                            opts.proxies.splice(proxies.index, 1);
+                            opts.proxy = proxies[proxies.index];
+                        }
                     }
-                }
 
-                if (err !== null && tries < opts.tries) {
-                    self.queueRequest(url, opts, callback, tries + 1);
+                    if (err !== null && tries < opts.tries) {
+                        self.queueRequest(url, opts, callback, tries + 1);
+
+                        if (self.opts.log === true) {
+                            self.command.error(err + ', retrying ' +
+                                            url.href + ' (' +
+                                            (tries + 1) + '/' +
+                                            opts.tries + ')');
+                        }
+                    } else {
+                        callback(err, res, data);
+                    }
 
+                    self.dequeueRequest();
+                    self.stack.pop();
+                })
+                .on('redirect', function (new_url) {
                     if (self.opts.log === true) {
-                        self.command.error(err + ', retrying ' +
-                                        url.href + ' (' +
-                                        (tries + 1) + '/' +
-                                        opts.tries + ')');
+                        self.command.log('[redirect] ' +
+                                         url.href + ' -> ' + new_url);
                     }
-                } else {
-                    callback(err, res, data);
-                }
 
-                self.dequeueRequest();
-                self.stack.pop();
-            })
-            .on('redirect', function (new_url) {
-                if (self.opts.log === true) {
-                    self.command.log('[redirect] ' +
-                                     url.href + ' -> ' + new_url);
-                }
-
-                url.href = new_url;
-            });
+                    url.href = new_url;
+                });
+    });
 };
 
 /**
@@ -367,6 +371,8 @@ Osmosis.prototype.resources = function () {
                 'requests: ' + this.requests +
                              ' (' + this.stack.requests + ' queued), ' +
 
+                'tokens: ' + parseInt(this.throttle.getTokensRemaining()) + ', ' +
+
                 'RAM: '      + toMB(mem.rss) + ' (' + memDiff + '), ' +
 
                 'libxml: '   + ((libxml_mem / mem.rss) * 100).toFixed(1) +

lib/commands/throttle.js

@@ -0,0 +1,22 @@
+/**
+ * Set a throttle. Short for `.config({ throttle: ... })`
+ *
+ * @function throttle
+ * @memberof Command
+ * @param {Number} tokensPerInterval Maximum number of tokens that can be
+ *  removed at any given moment and over the course of one interval.
+ * @param {String|Number} interval The interval length in milliseconds, or as
+ *  one of the following strings: 'second', 'minute', 'hour', day'.
+ * @see Osmosis.config
+ */
+
+var RateLimiter = require('limiter').RateLimiter;
+
+module.exports = function (tokensPerInterval, interval) {
+    this.instance.throttle = new RateLimiter(
+        tokensPerInterval || 1000,
+        interval || 1
+    );
+
+    return this;
+};

package.json

@@ -20,8 +20,9 @@
     "email": "[email protected]"
   },
   "dependencies": {
-    "needle": "1.0.0",
-    "libxmljs-dom": "0.0.7"
+    "libxmljs-dom": "0.0.7",
+    "limiter": "^1.1.0",
+    "needle": "1.0.0"
   },
   "devDependencies": {
     "jscs": ">=3.0.2",