Add ConcurrentMapper, MappingQueue, and WorkerQueue aliases

README.md

@@ -4,9 +4,11 @@
 
 `@shutterstock/p-map-iterable` provides several classes that allow processing results of `p-map`-style mapper functions by iterating the results as they are completed, with backpressure to limit the number of items that are processed ahead of the consumer.
 
+Use these classes for concurrent metadata lookups, capability probes, background status checks, or read/write pipelines. The [optional aliases](#optional-names-for-concurrent-work) `ConcurrentMapper`, `MappingQueue`, and `WorkerQueue` describe the input and result contracts without tying the work to a particular I/O operation.
+
 A common use case for `@shutterstock/p-map-iterable` is as a "prefetcher" that will fetch, for example, AWS S3 files in an AWS Lambda function. By prefetching large files the consumer is able to use 100% of the paid-for Lambda CPU time for the JS thread, rather than waiting idle while the next file is fetched. The backpressure (set by `maxUnread`) prevents the prefetcher from consuming unlimited memory or disk space by racing ahead of the consumer.
 
-These classes will typically be helpful in batch or queue consumers, not as much in request/response services.
+The caller supplies the operation to run for each input. Concurrency overlaps asynchronous work in the current JavaScript process; synchronous CPU work still uses the JavaScript event loop.
 
 # Example Usage Scenarios
 
@@ -25,14 +27,14 @@ for (const sourceId of sourceIds) {
 
 Each iteration takes 820ms total, but we waste time waiting for I/O. We could prefetch the next read (300ms) while processing (20ms) and writing (500ms), without changing the order of reads or writes.
 
-## Using `IterableMapper` as Prefetcher with Blocking Sequential Writes
+## Prefetching with `IterableMapper` and Blocking Sequential Writes
 
-`concurrency: 1` on the prefetcher preserves the order of the reads and and writes are sequential and blocking (unchanged).
+`concurrency: 1` on the prefetcher preserves the order of the reads, and writes remain sequential and blocking.
 
 ```typescript
 const source = new SomeSource();
 const sourceIds = [1, 2,... 1000];
-// Pre-reads up to 8 items serially and releases in sequential order
+// Prefetches serially within maxUnread and releases results in sequential order
 const sourcePrefetcher = new IterableMapper(sourceIds,
   async (sourceId) => source.read(sourceId),
   { concurrency: 1, maxUnread: 10 }
@@ -46,7 +48,7 @@ for await (const item of sourcePrefetcher) {    // may not block for fast source
 
 This reduces iteration time to 520ms by overlapping reads with processing/writing.
 
-## Using `IterableMapper` as Prefetcher with Background Sequential Writes with `IterableQueueMapperSimple`
+## Prefetching with `IterableMapper` and Background Sequential Writes with `IterableQueueMapperSimple`
 
 `concurrency: 1` on the prefetcher preserves the order of the reads.
 `concurrency: 1` on the flusher preserves the order of the writes, but allows the loop to iterate while last write is completing.
@@ -67,7 +69,7 @@ for await (const item of sourcePrefetcher) {    // may not block for fast source
   const outputItem = doSomeOperation(item);     // takes 20 ms of CPU
   await flusher.enqueue(outputItem);            // will periodically block for portion of write time
 }
-// Wait for all writes to complete
+// Close input permanently after the last enqueue and wait for all writes
 await flusher.onIdle();
 // Check for errors
 if (flusher.errors.length > 0) {
@@ -80,7 +82,7 @@ by overlapping reads and writes with the CPU processing step.
 In this contrived example, the loop time is reduced to 500ms - 20ms = 480ms.
 In cases where the CPU usage time is higher, the impact can be greater.
 
-## Using `IterableMapper` as Prefetcher with Out of Order Reads and Background Out of Order Writes with `IterableQueueMapperSimple`
+## Prefetching with `IterableMapper` and Out of Order Background Writes with `IterableQueueMapperSimple`
 
 For maximum throughput, allow out of order reads and writes with
 `IterableQueueMapper` (to iterate results with backpressure when too many unread items) or
@@ -102,7 +104,7 @@ for await (const item of sourcePrefetcher) {    // typically will not block
   const outputItem = doSomeOperation(item);     // takes 20 ms of CPU
   await flusher.enqueue(outputItem);            // typically will not block
 }
-// Wait for all writes to complete
+// Close input permanently after the last enqueue and wait for all writes
 await flusher.onIdle();
 // Check for errors
 if (flusher.errors.length > 0) {
@@ -152,20 +154,23 @@ These diagrams illustrate the differences in operation betweeen `p-map`, `p-queu
 # Features
 
 - [IterableMapper](https://tech.shutterstock.com/p-map-iterable/classes/IterableMapper.html)
+  - Also exported as `ConcurrentMapper`
   - Interface and concept based on: [p-map](https://github.com/sindresorhus/p-map)
   - Allows a sync or async iterable input
   - User supplied sync or async mapper function
   - Exposes an async iterable interface for consuming mapped items
   - Allows a maximum queue depth of mapped items - if the consumer stops consuming, the queue will fill up, at which point the mapper will stop being invoked until an item is consumed from the queue
   - This allows mapping with backpressure so that the mapper does not consume unlimited resources (e.g. memory, disk, network, event loop time) by racing ahead of the consumer
 - [IterableQueueMapper](https://tech.shutterstock.com/p-map-iterable/classes/IterableQueueMapper.html)
+  - Also exported as `MappingQueue`
   - Wraps `IterableMapper`
   - Adds items to the queue via the `enqueue` method
 - [IterableQueueMapperSimple](https://tech.shutterstock.com/p-map-iterable/classes/IterableQueueMapperSimple.html)
+  - Also exported as `WorkerQueue`
   - Wraps `IterableQueueMapper`
   - Discards results as they become available
   - Exposes any accumulated errors through the `errors` property instead of throwing an `AggregateError`
-  - Not actually `Iterable` - May rename this before 1.0.0
+  - Consumes results internally and exposes no result iterator
 
 ## Lower Level Utilities
 - [IterableQueue](https://tech.shutterstock.com/p-map-iterable/classes/IterableQueue.html)
@@ -184,6 +189,8 @@ See [p-map](https://github.com/sindresorhus/p-map) docs for a good start in unde
 
 The key difference between `IterableMapper` and `pMap` are that `IterableMapper` does not return when the entire mapping is done, rather it exposes an iterable that the caller loops through. This enables results to be processed while the mapping is still happening, while optionally allowing for backpressure to slow or stop the mapping if the caller is not consuming items fast enough. Common use cases include `prefetching` items from a remote service - the next set of requests are dispatched asyncronously while the current responses are processed and the prefetch requests will pause when the unread queue fills up.
 
+`ConcurrentMapper` is an alias for this class. It also fits general mapping work such as enriching project metadata or probing a list of services. Consume results with `for await`; with `concurrency` greater than one, results can arrive out of input order.
+
 See [examples/iterable-mapper.ts](./examples/iterable-mapper.ts) for an example.
 
 Run the example with `npm run example:iterable-mapper`
@@ -192,6 +199,8 @@ Run the example with `npm run example:iterable-mapper`
 
 `IterableQueueMapper` is similar to `IterableMapper` but instead of taking an iterable input it instead adds data via the `enqueue` method which will block if `maxUnread` will be reached by the current number of `mapper`'s running in parallel.
 
+`MappingQueue` is an alias for this class. Produce inputs and consume mapped results concurrently so the result buffer can drain. After the last awaited enqueue, call `done()` and finish consuming the iterator; `done()` closes input and does not wait for work to complete.
+
 See [examples/iterable-queue-mapper.ts](./examples/iterable-queue-mapper.ts) for an example.
 
 Run the example with `npm run example:iterable-queue-mapper`
@@ -200,14 +209,52 @@ Run the example with `npm run example:iterable-queue-mapper`
 
 `IterableQueueMapperSimple` is similar to `IterableQueueMapper` but instead exposing the results as an iterable it discards the results as soon as they are ready and exposes any errors through the `errors` property.
 
+`WorkerQueue` is an alias for this class. Supply one worker callback to process every enqueued input, such as a background Git status check. Await each enqueue for producer backpressure. After the last enqueue, await `onIdle()` to permanently close input and finish the accepted work, then inspect `errors`. An enqueue resolves when the input is accepted, rather than when its work completes.
+
 See [examples/iterable-queue-mapper-simple.ts](./examples/iterable-queue-mapper-simple.ts) for an example.
 
 Run the example with `npm run example:iterable-queue-mapper-simple`
 
+## Optional names for concurrent work
+
+The package exports three optional class aliases. Choose the name that makes the input and result contracts clearest in your application:
+
+| Alias | Original class | Input | Results and completion |
+| --- | --- | --- | --- |
+| `ConcurrentMapper` | `IterableMapper` | Sync or async iterable | Consume mapped results with `for await`. Mapping pauses when the unread result buffer fills. |
+| `MappingQueue` | `IterableQueueMapper` | Awaited `enqueue(item)` calls | Consume results concurrently with production. Call `done()` after the last enqueue, then finish consuming the iterator. |
+| `WorkerQueue` | `IterableQueueMapperSimple` | Awaited `enqueue(item)` calls | Results are consumed internally. After the last enqueue, await `onIdle()` and check `errors`. |
+
+```typescript
+import {
+  ConcurrentMapper,
+  MappingQueue,
+  WorkerQueue,
+} from '@shutterstock/p-map-iterable';
+```
+
+The aliases are the original classes, with the same constructor identity, generic instance types, and subclassing behavior. Existing imports continue to work. Defaults, result ordering, backpressure, and error handling are identical through either name.
+
+Matching option types are available from the package root using `import type`:
+
+| Option alias | Original option type | Configuration |
+| --- | --- | --- |
+| `ConcurrentMapperOptions` | `IterableMapperOptions` | `concurrency`, `maxUnread`, `stopOnMapperError` |
+| `MappingQueueOptions` | `IterableQueueMapperOptions` | `concurrency`, `maxUnread`, `stopOnMapperError` |
+| `WorkerQueueOptions` | `IterableQueueMapperSimpleOptions` | `concurrency` |
+
+`MappingQueue` must have a result consumer even when you do not need the results. Awaiting every enqueue before starting iteration can block once the result buffer fills. Use `WorkerQueue` when results can be discarded. Its `onIdle()` permanently ends input; subsequent enqueues reject. Worker failures are collected in `errors` while other inputs continue to run, so check that property after shutdown.
+
+Each queue uses one fixed callback supplied at construction. Awaiting `enqueue()` provides producer backpressure and confirms acceptance of an input. Calling it without awaiting can accumulate pending inputs. For ongoing event-driven work, the application supplies any admission limits, cancellation, deduplication, or per-item completion handles. `WorkerQueue.onIdle()` is a final shutdown operation, so an application that needs reusable idle waits must manage that separately.
+
+See [examples/semantic-aliases.ts](./examples/semantic-aliases.ts) for metadata enrichment, queued capability probes with a concurrent result consumer, and background status checks with collected errors. Run it with `npm run example:semantic-aliases`.
+
 # Contributing - Setting up Build Environment
 
 - `nvm use`
 - `npm i`
 - `npm run build`
+- `npm run build:docs`
 - `npm run lint`
 - `npm run test`
+- `npm run example:semantic-aliases`

examples/iterable-mapper.ts

@@ -41,7 +41,7 @@ async function main() {
   let queuedButUnreadFileSizeGB = 0;
   let callCount = 0;
 
-  // Create an item prefetcher with IterableMapper
+  // Create a prefetching mapper with IterableMapper (also exported as ConcurrentMapper)
   // Use `npm i it-batch`'s `batch` function to batch the source iterable
   // into batches of a specific size, such as 50, if making batch requests
   // to a remote service.

examples/iterable-queue-mapper-simple.ts

@@ -30,16 +30,16 @@ async function main() {
   let total = 0;
   let callCount = 0;
 
-  // Create an item processor with IterableQueueMapperSimple
-  const backgroundFlusher = new IterableQueueMapperSimple(
+  // Create an item processor with IterableQueueMapperSimple (also exported as WorkerQueue)
+  const workers = new IterableQueueMapperSimple(
     // mapper function
     async (value: number): Promise<void> => {
       const myCallCount = callCount++;
       total += value;
 
       console.log(`Mapper Call Start ${myCallCount}, Value: ${value}, Total: ${total}`);
 
-      // Simulate flushing an async item with varied delays
+      // Simulate asynchronous background work with varied delays
       await sleep(Math.random() * 10000);
 
       if (value % 5 === 0) {
@@ -51,27 +51,27 @@ async function main() {
     { concurrency: 3 },
   );
 
-  // Add items to be flushed to the queue in the background
+  // Add inputs for the fixed worker callback
   // This will pause when the queue is full and resume when there is capacity
   const jobAdder = (async () => {
     for await (const item of iterator) {
       console.log(`Enqueue Start ${item}`);
-      await backgroundFlusher.enqueue(item);
+      await workers.enqueue(item);
       console.log(`Enqueue Done  ${item}`);
     }
   })();
 
   // Wait for the job adder to finish adding the jobs
-  // (it's throughput is constrained by the flushers's concurrency)
+  // (its throughput is constrained by the worker concurrency)
   await jobAdder;
 
-  // Wait for the async flusher to finish flushing all items
-  await backgroundFlusher.onIdle();
+  // Close input permanently and wait for all accepted work
+  await workers.onIdle();
 
   // Check for errors
-  if (backgroundFlusher.errors.length > 0) {
+  if (workers.errors.length > 0) {
     console.error('Errors:');
-    backgroundFlusher.errors.forEach(({ error, item }) =>
+    workers.errors.forEach(({ error, item }) =>
       console.error(
         `${item} had error: ${(error as Error).message ? (error as Error).message : error}`,
       ),

examples/iterable-queue-mapper.ts

@@ -43,7 +43,7 @@ async function main() {
   let queuedButUnreadFileSizeGB = 0;
   let callCount = 0;
 
-  // Create an item prefetcher with IterableQueueMapper
+  // Create a queued prefetching mapper with IterableQueueMapper (also exported as MappingQueue)
   const prefetcher = new IterableQueueMapper(
     async (value: {
       fileName: string;
@@ -76,7 +76,7 @@ async function main() {
 
   let loopCount = 0;
 
-  // Add items to the queue in the background
+  // Add inputs in the background while the consumer below drains the result buffer
   const jobAdder = (async () => {
     for await (const item of iterator) {
       console.log(`Enqueue Start FileName: ${item.fileName}, FileSizeGB: ${item.fileSizeGB}`);
@@ -131,7 +131,7 @@ async function main() {
   }
 
   // Wait for the job adder to finish adding the jobs
-  // (it's throughput is constrained by the prefetcher's concurrency)
+  // (its throughput is constrained by the mapper concurrency)
   await jobAdder;
 
   console.log(`QueuedButUnreadFileSizeGB: ${queuedButUnreadFileSizeGB}`);

examples/semantic-aliases.ts

@@ -0,0 +1,82 @@
+/* eslint-disable no-console */
+import { ConcurrentMapper, MappingQueue, WorkerQueue } from '@shutterstock/p-map-iterable';
+
+async function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function concurrentMapperExample(): Promise<void> {
+  const mapper = new ConcurrentMapper(
+    [1, 2, 3, 4, 5],
+    async (id) => {
+      await delay(10); // Enrich each input with metadata from another service.
+      return { id, label: `Project ${id}` };
+    },
+    { concurrency: 2, maxUnread: 3 },
+  );
+
+  for await (const item of mapper) {
+    console.log(`Metadata ${item.id}: ${item.label}`);
+    await delay(5); // Process a result while other mapper calls run concurrently.
+  }
+}
+
+async function mappingQueueExample(): Promise<void> {
+  const queue = new MappingQueue(
+    async (id: number) => {
+      await delay(10); // Probe a worker and return its capabilities.
+      return { id, available: true };
+    },
+    { concurrency: 2, maxUnread: 3 },
+  );
+
+  // Produce more items than maxUnread. A concurrent consumer is required to
+  // release backpressure; waiting for all enqueues before iterating would block.
+  const producer = (async () => {
+    for (let id = 1; id <= 20; id++) {
+      await queue.enqueue(id);
+    }
+    queue.done();
+  })();
+
+  const consumer = (async () => {
+    for await (const result of queue) {
+      console.log(`Worker ${result.id}: available=${result.available}`);
+    }
+  })();
+
+  await Promise.all([producer, consumer]);
+}
+
+async function workerQueueExample(): Promise<void> {
+  const queue = new WorkerQueue(
+    async (id: number) => {
+      await delay(10); // Run a background status check with no returned result.
+      if (id === 3) throw new Error('Simulated task failure');
+      console.log(`Status check ${id}: complete`);
+    },
+    { concurrency: 2 },
+  );
+
+  for (let id = 1; id <= 5; id++) {
+    await queue.enqueue(id);
+  }
+
+  // onIdle() closes input permanently and waits for the accepted work.
+  // Mapper failures are collected in errors instead of rejecting onIdle().
+  await queue.onIdle();
+  for (const { item, error } of queue.errors) {
+    console.log(`Status check ${item}: failed`, error);
+  }
+}
+
+async function main(): Promise<void> {
+  await concurrentMapperExample();
+  await mappingQueueExample();
+  await workerQueueExample();
+}
+
+void main().catch((error: unknown) => {
+  console.error(error);
+  process.exitCode = 1;
+});

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@shutterstock/p-map-iterable",
   "version": "0.0.0",
-  "description": "Set of classes used for async prefetching with backpressure (IterableMapper) and async flushing with backpressure (IterableQueueMapper, IterableQueueMapperSimple)",
+  "description": "Concurrent mapping and queued background work with backpressure, streaming results, and collected errors",
   "main": "dist/src/index.js",
   "types": "dist/src/index.d.ts",
   "publishConfig": {
@@ -29,6 +29,7 @@
     "example:iterable-mapper": "ts-node -r tsconfig-paths/register examples/iterable-mapper.ts",
     "example:iterable-queue-mapper": "ts-node -r tsconfig-paths/register examples/iterable-queue-mapper.ts",
     "example:iterable-queue-mapper-simple": "ts-node -r tsconfig-paths/register examples/iterable-queue-mapper-simple.ts",
+    "example:semantic-aliases": "ts-node -r tsconfig-paths/register examples/semantic-aliases.ts",
     "test": "jest",
     "lint": "eslint .",
     "lint-and-fix": "eslint . --fix"