@rushstack/node-core-library - npm Package Compare versions

Comparing version 4.1.0 to 4.2.0

lib/Async.d.ts

@@ -5,3 +5,4 @@ /**
  * @remarks
- * Used with {@link Async.mapAsync} and {@link Async.forEachAsync}.
+ * Used with {@link (Async:class).(mapAsync:1)}, {@link (Async:class).(mapAsync:2)} and
+ * {@link (Async:class).(forEachAsync:1)}, and {@link (Async:class).(forEachAsync:2)}.
  *
@@ -12,6 +13,12 @@ * @public
     /**
-     * Optionally used with the  {@link Async.mapAsync} and {@link Async.forEachAsync}
-     * to limit the maximum number of concurrent promises to the specified number.
+     * Optionally used with the  {@link (Async:class).(mapAsync:1)}, {@link (Async:class).(mapAsync:2)} and
+     * {@link (Async:class).(forEachAsync:1)}, and {@link (Async:class).(forEachAsync:2)} to limit the maximum
+     * number of concurrent promises to the specified number.
      */
     concurrency?: number;
+    /**
+     * Optionally used with the {@link (Async:class).(forEachAsync:2)} to enable weighted operations where an operation can
+     * take up more or less than one concurrency unit.
+     */
+    weighted?: boolean;
 }
@@ -30,2 +37,15 @@ /**
 /**
+ * @remarks
+ * Used with {@link (Async:class).(forEachAsync:2)} and {@link (Async:class).(mapAsync:2)}.
+ *
+ * @public
+ */
+export interface IWeighted {
+    /**
+     * The weight of the element, used to determine the concurrency units that it will take up.
+     *  Must be a whole number greater than or equal to 0.
+     */
+    weight: number;
+}
+/**
  * Utilities for parallel asynchronous operations, for use with the system `Promise` APIs.
@@ -56,5 +76,32 @@ *
      */
-    static mapAsync<TEntry, TRetVal>(iterable: Iterable<TEntry> | AsyncIterable<TEntry>, callback: (entry: TEntry, arrayIndex: number) => Promise<TRetVal>, options?: IAsyncParallelismOptions | undefined): Promise<TRetVal[]>;
+    static mapAsync<TEntry, TRetVal>(iterable: Iterable<TEntry> | AsyncIterable<TEntry>, callback: (entry: TEntry, arrayIndex: number) => Promise<TRetVal>, options?: (IAsyncParallelismOptions & {
+        weighted?: false;
+    }) | undefined): Promise<TRetVal[]>;
     /**
      * Given an input array and a `callback` function, invoke the callback to start a
+     * promise for each element in the array.  Returns an array containing the results.
+     *
+     * @remarks
+     * This API is similar to the system `Array#map`, except that the loop is asynchronous,
+     * and the maximum number of concurrent units can be throttled
+     * using {@link IAsyncParallelismOptions.concurrency}. Using the {@link IAsyncParallelismOptions.weighted}
+     * option, the weight of each operation can be specified, which determines how many concurrent units it takes up.
+     *
+     * If `callback` throws a synchronous exception, or if it returns a promise that rejects,
+     * then the loop stops immediately.  Any remaining array items will be skipped, and
+     * overall operation will reject with the first error that was encountered.
+     *
+     * @param iterable - the array of inputs for the callback function
+     * @param callback - a function that starts an asynchronous promise for an element
+     *   from the array
+     * @param options - options for customizing the control flow
+     * @returns an array containing the result for each callback, in the same order
+     *   as the original input `array`
+     */
+    static mapAsync<TEntry extends IWeighted, TRetVal>(iterable: Iterable<TEntry> | AsyncIterable<TEntry>, callback: (entry: TEntry, arrayIndex: number) => Promise<TRetVal>, options: IAsyncParallelismOptions & {
+        weighted: true;
+    }): Promise<TRetVal[]>;
+    private static _forEachWeightedAsync;
+    /**
+     * Given an input array and a `callback` function, invoke the callback to start a
      * promise for each element in the array.
@@ -76,4 +123,34 @@ *
      */
-    static forEachAsync<TEntry>(iterable: Iterable<TEntry> | AsyncIterable<TEntry>, callback: (entry: TEntry, arrayIndex: number) => Promise<void>, options?: IAsyncParallelismOptions | undefined): Promise<void>;
+    static forEachAsync<TEntry>(iterable: Iterable<TEntry> | AsyncIterable<TEntry>, callback: (entry: TEntry, arrayIndex: number) => Promise<void>, options?: (IAsyncParallelismOptions & {
+        weighted?: false;
+    }) | undefined): Promise<void>;
     /**
+     * Given an input array and a `callback` function, invoke the callback to start a
+     * promise for each element in the array.
+     *
+     * @remarks
+     * This API is similar to the other `Array#forEachAsync`, except that each item can have
+     * a weight that determines how many concurrent operations are allowed. The unweighted
+     * `Array#forEachAsync` is a special case of this method where weight = 1 for all items.
+     *
+     * The maximum number of concurrent operations can still be throttled using
+     * {@link IAsyncParallelismOptions.concurrency}, however it no longer determines the
+     * maximum number of operations that can be in progress at once. Instead, it determines the
+     * number of concurrency units that can be in progress at once. The weight of each operation
+     * determines how many concurrency units it takes up. For example, if the concurrency is 2
+     * and the first operation has a weight of 2, then only one more operation can be in progress.
+     *
+     * If `callback` throws a synchronous exception, or if it returns a promise that rejects,
+     * then the loop stops immediately.  Any remaining array items will be skipped, and
+     * overall operation will reject with the first error that was encountered.
+     *
+     * @param iterable - the array of inputs for the callback function
+     * @param callback - a function that starts an asynchronous promise for an element
+     *   from the array
+     * @param options - options for customizing the control flow
+     */
+    static forEachAsync<TEntry extends IWeighted>(iterable: Iterable<TEntry> | AsyncIterable<TEntry>, callback: (entry: TEntry, arrayIndex: number) => Promise<void>, options: IAsyncParallelismOptions & {
+        weighted: true;
+    }): Promise<void>;
+    /**
      * Return a promise that resolves after the specified number of milliseconds.
@@ -87,2 +164,7 @@ */
     /**
+     * Ensures that the argument is a valid {@link IWeighted}, with a `weight` argument that
+     * is a positive integer or 0.
+     */
+    static validateWeightedIterable(operation: IWeighted): void;
+    /**
      * Returns a Signal, a.k.a. a "deferred promise".
@@ -89,0 +171,0 @@ */

lib/Async.js

@@ -19,2 +19,18 @@ "use strict";
 exports.AsyncQueue = exports.Async = void 0;
+function toWeightedIterator(iterable, useWeights) {
+    const iterator = (iterable[Symbol.iterator] ||
+        iterable[Symbol.asyncIterator]).call(iterable);
+    return {
+        [Symbol.asyncIterator]: () => ({
+            next: async () => {
+                // The await is necessary here, but TS will complain - it's a false positive.
+                const { value, done } = await iterator.next();
+                return {
+                    value: { element: value, weight: useWeights ? value === null || value === void 0 ? void 0 : value.weight : 1 },
+                    done: !!done
+                };
+            }
+        })
+    };
+}
 /**
@@ -26,24 +42,5 @@ * Utilities for parallel asynchronous operations, for use with the system `Promise` APIs.
 class Async {
-    /**
-     * Given an input array and a `callback` function, invoke the callback to start a
-     * promise for each element in the array.  Returns an array containing the results.
-     *
-     * @remarks
-     * This API is similar to the system `Array#map`, except that the loop is asynchronous,
-     * and the maximum number of concurrent promises can be throttled
-     * using {@link IAsyncParallelismOptions.concurrency}.
-     *
-     * If `callback` throws a synchronous exception, or if it returns a promise that rejects,
-     * then the loop stops immediately.  Any remaining array items will be skipped, and
-     * overall operation will reject with the first error that was encountered.
-     *
-     * @param iterable - the array of inputs for the callback function
-     * @param callback - a function that starts an asynchronous promise for an element
-     *   from the array
-     * @param options - options for customizing the control flow
-     * @returns an array containing the result for each callback, in the same order
-     *   as the original input `array`
-     */
     static async mapAsync(iterable, callback, options) {
         const result = [];
+        // @ts-expect-error https://github.com/microsoft/TypeScript/issues/22609, it succeeds against the implementation but fails against the overloads
         await Async.forEachAsync(iterable, async (item, arrayIndex) => {
@@ -54,24 +51,6 @@ result[arrayIndex] = await callback(item, arrayIndex);
     }
-    /**
-     * Given an input array and a `callback` function, invoke the callback to start a
-     * promise for each element in the array.
-     *
-     * @remarks
-     * This API is similar to the system `Array#forEach`, except that the loop is asynchronous,
-     * and the maximum number of concurrent promises can be throttled
-     * using {@link IAsyncParallelismOptions.concurrency}.
-     *
-     * If `callback` throws a synchronous exception, or if it returns a promise that rejects,
-     * then the loop stops immediately.  Any remaining array items will be skipped, and
-     * overall operation will reject with the first error that was encountered.
-     *
-     * @param iterable - the array of inputs for the callback function
-     * @param callback - a function that starts an asynchronous promise for an element
-     *   from the array
-     * @param options - options for customizing the control flow
-     */
-    static async forEachAsync(iterable, callback, options) {
+    static async _forEachWeightedAsync(iterable, callback, options) {
         await new Promise((resolve, reject) => {
             const concurrency = (options === null || options === void 0 ? void 0 : options.concurrency) && options.concurrency > 0 ? options.concurrency : Infinity;
-            let operationsInProgress = 0;
+            let concurrentUnitsInProgress = 0;
             const iterator = (iterable[Symbol.iterator] ||
@@ -83,6 +62,8 @@ iterable[Symbol.asyncIterator]).call(iterable);
             async function queueOperationsAsync() {
-                while (operationsInProgress < concurrency && !iteratorIsComplete && !promiseHasResolvedOrRejected) {
+                while (concurrentUnitsInProgress < concurrency &&
+                    !iteratorIsComplete &&
+                    !promiseHasResolvedOrRejected) {
                     // Increment the concurrency while waiting for the iterator.
                     // This function is reentrant, so this ensures that at most `concurrency` executions are waiting
-                    operationsInProgress++;
+                    concurrentUnitsInProgress++;
                     const currentIteratorResult = await iterator.next();
@@ -92,5 +73,12 @@ // eslint-disable-next-line require-atomic-updates
                     if (!iteratorIsComplete) {
-                        Promise.resolve(callback(currentIteratorResult.value, arrayIndex++))
+                        const currentIteratorValue = currentIteratorResult.value;
+                        Async.validateWeightedIterable(currentIteratorValue);
+                        const weight = Math.min(currentIteratorValue.weight, concurrency);
+                        // If it's a weighted operation then add the rest of the weight, removing concurrent units if weight < 1.
+                        // Cap it to the concurrency limit, otherwise higher weights can cause issues in the case where 0 weighted
+                        // operations are present.
+                        concurrentUnitsInProgress += weight - 1;
+                        Promise.resolve(callback(currentIteratorValue.element, arrayIndex++))
                             .then(async () => {
-                            operationsInProgress--;
+                            concurrentUnitsInProgress -= weight;
                             await onOperationCompletionAsync();
@@ -105,3 +93,3 @@ })
                         // The iterator is complete and there wasn't a value, so untrack the waiting state.
-                        operationsInProgress--;
+                        concurrentUnitsInProgress--;
                     }
@@ -115,3 +103,3 @@ }
                 if (!promiseHasResolvedOrRejected) {
-                    if (operationsInProgress === 0 && iteratorIsComplete) {
+                    if (concurrentUnitsInProgress === 0 && iteratorIsComplete) {
                         promiseHasResolvedOrRejected = true;
@@ -131,2 +119,5 @@ resolve();
     }
+    static async forEachAsync(iterable, callback, options) {
+        await Async._forEachWeightedAsync(toWeightedIterator(iterable, options === null || options === void 0 ? void 0 : options.weighted), callback, options);
+    }
     /**
@@ -161,2 +152,14 @@ * Return a promise that resolves after the specified number of milliseconds.
     /**
+     * Ensures that the argument is a valid {@link IWeighted}, with a `weight` argument that
+     * is a positive integer or 0.
+     */
+    static validateWeightedIterable(operation) {
+        if (operation.weight < 0) {
+            throw new Error('Weight must be a whole number greater than or equal to 0');
+        }
+        if (operation.weight % 1 !== 0) {
+            throw new Error('Weight must be a whole number greater than or equal to 0');
+        }
+    }
+    /**
      * Returns a Signal, a.k.a. a "deferred promise".
@@ -163,0 +166,0 @@ */

lib/index.d.ts

@@ -7,3 +7,3 @@ /**
 export { AlreadyReportedError } from './AlreadyReportedError';
-export { Async, AsyncQueue, IAsyncParallelismOptions, IRunWithRetriesOptions } from './Async';
+export { Async, AsyncQueue, IAsyncParallelismOptions, IRunWithRetriesOptions, IWeighted } from './Async';
 export { Brand } from './PrimitiveTypes';
@@ -10,0 +10,0 @@ export { FileConstants, FolderConstants } from './Constants';

package.json

 {
   "name": "@rushstack/node-core-library",
-  "version": "4.1.0",
+  "version": "4.2.0",
   "description": "Core libraries that every NodeJS toolchain project should use",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

1046828

increased by1.79%

Lines of code

13514

increased by1.25%

No dependency changes