1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
import TokenBucket from './internal/TokenBucket';
import noop from 'lodash/noop';
import once from './once';
import iterator from './iterator';
import onlyOnce from './onlyOnce';
/**
* The same as [`eachOf`]{@link module:Collections.eachOf} but runs a maximum of `limit` async operations at a
* time.
*
* @name eachOfRateLimit
* @static
* @memberOf module:Collections
* @method
* @see [async.eachOf]{@link module:Collections.eachOf}
* @alias forEachOfRateLimit
* @category Collection
* @param {Array|Iterable|Object} coll - A collection to iterate over.
* @param {number} limit - The maximum number of async operations at a time.
* @param {Function} iteratee - A function to apply to each
* item in `coll`. The `key` is the item's key, or index in the case of an
* array. The iteratee is passed a `callback(err)` which must be called once it
* has completed. If no error has occurred, the callback should be run without
* arguments or with an explicit `null` argument. Invoked with
* (item, key, callback).
* @param {Object} rateLimitOptions - the rate limiting options. options.bucketSize
* will limit the amount of items queued within options.interval (miliseconds).
* @param {Function} [callback] - A callback which is called when all
* `iteratee` functions have finished, or an error occurs. Invoked with (err).
*/
export default function(coll, iteratee, options, callback) {
callback = once(callback || noop);
var tokenBucket = new TokenBucket(options.bucketSize, options.interval);
function iterateeCallback(err) {
if (err) {
tokenBucket.empty();
callback(err);
}
// check nextElem iterator is exhausted (elem == null) to be sure
// we don't exit immediately due to a synchronous iteratee
else if (tokenBucket.queued === 0 && elem === null) {
return callback(null);
}
}
function enqueue(value, key) {
tokenBucket.enqueue(function() {
iteratee(value, key, onlyOnce(iterateeCallback));
});
}
var nextElem = iterator(coll);
var elem;
while((elem = nextElem()) !== null) {
enqueue(elem.value, elem.key);
}
}
|
