limit.js (4634B)
1 /** 2 * @license Apache-2.0 3 * 4 * Copyright (c) 2018 The Stdlib Authors. 5 * 6 * Licensed under the Apache License, Version 2.0 (the "License"); 7 * you may not use this file except in compliance with the License. 8 * You may obtain a copy of the License at 9 * 10 * http://www.apache.org/licenses/LICENSE-2.0 11 * 12 * Unless required by applicable law or agreed to in writing, software 13 * distributed under the License is distributed on an "AS IS" BASIS, 14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15 * See the License for the specific language governing permissions and 16 * limitations under the License. 17 */ 18 19 'use strict'; 20 21 // MODULES // 22 23 var logger = require( 'debug' ); 24 var hasOwnProp = require( '@stdlib/assert/has-own-property' ); 25 26 27 // VARIABLES // 28 29 var debug = logger( 'group-by-async:limit' ); 30 31 32 // MAIN // 33 34 /** 35 * Invokes an indicator function once for each element in a collection, limiting the number of concurrently pending functions. 36 * 37 * ## Notes 38 * 39 * - We need to cache the collection value to prevent the edge case where, during the invocation of the indicator function, the element at index `i` is swapped for some other value. For some, that might be a feature; here, we take the stance that one should be less clever. 40 * - Checking for an "own" property is necessary to guard against the edge case where an indicator function returns a group identifier which matches a method or property on the `Object` prototype. 41 * 42 * 43 * @private 44 * @param {Collection} collection - input collection 45 * @param {Options} opts - function options 46 * @param {*} [opts.thisArg] - execution context 47 * @param {PositiveInteger} [opts.limit] - maximum number of pending function invocations 48 * @param {string} [options.returns] - output format 49 * @param {Function} indicator - indicator function 50 * @param {Callback} done - function to invoke upon completion or upon encountering an error 51 * @returns {void} 52 */ 53 function limit( collection, opts, indicator, done ) { 54 var maxIndex; 55 var count; 56 var flg; 57 var lim; 58 var len; 59 var idx; 60 var out; 61 var i; 62 63 len = collection.length; 64 debug( 'Collection length: %d', len ); 65 66 out = {}; 67 if ( len === 0 ) { 68 debug( 'Finished processing a collection.' ); 69 return done( null, out ); 70 } 71 if ( len < opts.limit ) { 72 lim = len; 73 } else { 74 lim = opts.limit; 75 } 76 debug( 'Concurrency limit: %d', lim ); 77 debug( 'Number of arguments: %d', indicator.length ); 78 79 maxIndex = len - 1; 80 count = 0; 81 idx = -1; 82 for ( i = 0; i < lim; i++ ) { 83 // This guard is necessary to protect against synchronous functions which exhaust all collection elements... 84 if ( idx < maxIndex ) { 85 next(); // eslint-disable-line callback-return 86 } 87 } 88 /** 89 * Callback to invoke a provided function for the next element in a collection. 90 * 91 * @private 92 */ 93 function next() { 94 var v; 95 var j; 96 97 idx += 1; 98 j = idx; 99 v = collection[ j ]; 100 101 debug( 'Collection element %d: %s.', j, JSON.stringify( v ) ); 102 if ( indicator.length === 2 ) { 103 indicator.call( opts.thisArg, v, cb ); 104 } else if ( indicator.length === 3 ) { 105 indicator.call( opts.thisArg, v, j, cb ); 106 } else { 107 indicator.call( opts.thisArg, v, j, collection, cb ); 108 } 109 /** 110 * Callback invoked once a provided function finishes processing a collection element. 111 * 112 * @private 113 * @param {*} [error] - error 114 * @param {*} [group] - group 115 * @returns {void} 116 */ 117 function cb( error, group ) { 118 if ( flg ) { 119 // Prevent further processing of collection elements: 120 return; 121 } 122 if ( error ) { 123 flg = true; 124 return clbk( error ); 125 } 126 debug( 'Collection element %d group: %s.', j, group ); 127 128 // Default is to return values... 129 if ( opts.returns === 'indices' ) { 130 if ( hasOwnProp( out, group ) ) { 131 out[ group ].push( j ); 132 } else { 133 out[ group ] = [ j ]; 134 } 135 } else if ( opts.returns === '*' ) { 136 if ( hasOwnProp( out, group ) ) { 137 out[ group ].push( [ j, v ] ); 138 } else { 139 out[ group ] = [ [ j, v ] ]; 140 } 141 } else if ( hasOwnProp( out, group ) ) { 142 out[ group ].push( v ); 143 } else { 144 out[ group ] = [ v ]; 145 } 146 clbk(); 147 } 148 } 149 150 /** 151 * Callback invoked once ready to process the next collection element. 152 * 153 * @private 154 * @param {*} [error] - error 155 * @returns {void} 156 */ 157 function clbk( error ) { 158 if ( error ) { 159 debug( 'Encountered an error: %s', error.message ); 160 return done( error ); 161 } 162 count += 1; 163 debug( 'Processed %d of %d collection elements.', count, len ); 164 if ( idx < maxIndex ) { 165 return next(); 166 } 167 if ( count === len ) { 168 debug( 'Finished processing a collection.' ); 169 return done( null, out ); 170 } 171 } 172 } 173 174 175 // EXPORTS // 176 177 module.exports = limit;