Async generator module

A comprehensive utility library for working with asynchronous generators in TypeScript. This package provides a rich set of composable functions that simplify complex asynchronous data processing workflows, following functional programming principles.

This library empowers developers to efficiently handle streaming data, implement backpressure, process items concurrently, and transform async data flows with minimal boilerplate. From basic operations like mapping, filtering, and reducing to advanced patterns like batching, windowing, and controlled concurrency, @prelude/async-generator offers a consistent API that makes working with asynchronous sequences as intuitive as working with arrays. The fully typed implementation ensures type safety throughout your async processing pipelines.

Installation

npm i -E @prelude/async-generator
Copy

Usage

import * as G from '@prelude/async-generator'
Copy

API Reference

• append: Appends values from an iterable after the original async iterable values
• areUnique: Checks if all values in an async iterable are unique
• array: Collects all values from an async iterable into an array
• assign: Mutates objects by assigning properties from a function result
• at: Retrieves a value at a specific index from an async iterable
• batch: Groups values from an async iterable into fixed-size batches
• bigproduct: Computes the product of all values as a BigInt
• bigsum: Computes the sum of all values as a BigInt
• buffered: Buffers values from an async iterable for controlled processing
• cargo: Accumulates values into dynamic batches based on processing timing
• cartesianProduct: Generates all possible combinations from multiple iterables
• compact: Filters out null and undefined values from an async iterable
• concat: Combines multiple async iterables sequentially into a single async generator
• consume: Processes all values from an async iterable for side effects
• count: Counts elements in an async iterable that satisfy a predicate
• cycle: Repeats values from an async iterable a specified number of times
• defined: Similar to compact but with TypeScript type narrowing
• every: Checks if all values in an async iterable satisfy a predicate
• filter: Creates a transformer that only yields values passing a predicate test
• jitter: Adds random time delays between yielded values
• map: Applies a mapping function to each value, with support for concurrency
• ofInterval: Creates an async generator that yields timestamps at regular intervals
• ofIterable: Converts any synchronous iterable into an async generator
• ofNext: Creates an async generator based on a provided next function
• pipe: Composes functions from left to right for serial transformations
• prelude: Core utility functions and types for the library
• prepend: Yields values from an iterable before the original async iterable values
• range: Creates an async generator yielding a sequence of numbers
• reduce: Reduces values from an async iterable to a single result
• skip: Skips the first n values from an async iterable and yields the rest
• sleep: Utility to pause execution for a specified duration
• some: Checks if at least one value in an async iterable satisfies a predicate
• step: Yields every nth value from an async iterable
• sum: Computes the sum of all values in an async iterable
• take: Limits the number of values yielded from an async iterable
• tap: Applies a side-effect function to each value without changing them
• unwrapIndexed: Restores original order of values based on their indices
• window: Creates sliding windows of values from an async iterable
• withIndex: Pairs each value with its sequential index in an object
• yield: Creates an async generator that yields a single value once

Cancellation

sleep, jitter, ofInterval, map, tap and consume accept signal in their options. Aborting makes the generator throw (or consume reject) with signal.reason, clears timers and stops pulling from the source, which is returned. map/tap throw at once — concurrent ones fail their output and close their input, serial ones race the pending pull and mapping against the signal — and drop the results of mappings still in flight; consume awaits callbacks already in flight before rejecting.

const controller = new AbortController()
await G.pipe(
  source,
  G.map(fetchOne, { concurrency: 4, signal: controller.signal }),
  G.consume(store, { signal: controller.signal })
)
Copy

License

This package is dedicated to the public domain under CC0 1.0.