@studnicky/concurrency

Keyed async channels, counting semaphores, concurrent-call coalescing, and async iterable combinators.

Install

bash

pnpm add @studnicky/concurrency

Requires @studnicky:registry=https://npm.pkg.github.com in .npmrc.

Usage

Channel and Semaphore

Channel provides keyed producer/consumer buffering; Semaphore gates concurrent access to a shared resource with a counting permit:

import { Channel, Semaphore } from '../src/index.js';

class ChannelSemaphoreDemo {
  static async runChannel(): Promise<void> {
    const channel = Channel.create<number>();

    // Publish items before subscribing — they buffer in the channel
    await channel.publish('nums', 1);
    await channel.publish('nums', 2);
    await channel.publish('nums', 3);
    await channel.close();

    const received: number[] = [];
    for await (const item of channel.subscribe('nums')) {
      received.push(item);
    }

    console.log('Channel received:', received);
    assert.deepEqual(received, [1, 2, 3]);
  }

  static async runChannelMultiKey(): Promise<void> {
    const channel = Channel.create<string>();

    // Two independent keys — no cross-talk
    await channel.publish('a', 'alpha');
    await channel.publish('b', 'beta');
    await channel.publish('a', 'apple');
    await channel.close();

    const fromA: string[] = [];
    for await (const item of channel.subscribe('a')) {
      fromA.push(item);
    }

    const fromB: string[] = [];
    for await (const item of channel.subscribe('b')) {
      fromB.push(item);
    }

    console.log('Channel key=a:', fromA);
    console.log('Channel key=b:', fromB);
    assert.deepEqual(fromA, ['alpha', 'apple']);
    assert.deepEqual(fromB, ['beta']);
  }

  static async runChannelConcurrent(): Promise<void> {
    const channel = Channel.create<number>();
    const received: number[] = [];

    // Start subscriber before publishing — it will wait for items
    const consuming = (async () => {
      for await (const item of channel.subscribe('live')) {
        received.push(item);
      }
    })();

    // Publish asynchronously, then close
    await Promise.resolve();
    await channel.publish('live', 10);
    await channel.publish('live', 20);
    await channel.close();

    await consuming;

    console.log('Channel concurrent received:', received);
    assert.deepEqual(received, [10, 20]);
  }

  static async runSemaphoreWithPermit(): Promise<void> {
    const sem = Semaphore.create({ 'permits': 2 });

    console.log('Semaphore permits:', sem.permits, 'available:', sem.available);

    let concurrent = 0;
    let maxConcurrent = 0;

    const task = (): Promise<void> =>
    { const result = sem.withPermit(async () => {
      concurrent += 1;
      maxConcurrent = Math.max(maxConcurrent, concurrent);
      // Yield to the event loop so tasks interleave
      await new Promise<void>((resolve) => { setImmediate(resolve); });
      concurrent -= 1;
    }); return result; };

    // Launch 4 tasks against a semaphore of 2
    await Promise.all([task(), task(), task(), task()]);

    console.log('Semaphore maxConcurrent:', maxConcurrent, '(≤ 2), available after:', sem.available);
    assert.ok(maxConcurrent <= 2, `maxConcurrent ${maxConcurrent} exceeded permits`);
  }

  static async runSemaphoreAcquireRelease(): Promise<void> {
    const sem = Semaphore.create({ 'permits': 1 });

    const release = await sem.acquire();
    const afterAcquire = sem.available;
    console.log('Semaphore available after acquire:', afterAcquire);

    await release();
    const afterRelease = sem.available;
    console.log('Semaphore available after release:', afterRelease);
    assert.equal(afterAcquire, 0);
    assert.equal(afterRelease, 1);
  }
}

Coalesce: deduplicate concurrent calls by key

All concurrent callers for the same key share a single in-flight promise; sequential callers each invoke the factory independently:

import { Coalesce } from '../src/index.js';

class CoalesceDemo {
  static async runSharedInFlight(): Promise<void> {
    const coalesce = Coalesce.create<string>();
    let callCount = 0;

    const factory = (): Promise<string> => {
      callCount += 1;
      return new Promise<string>((resolve) => {
        setImmediate(() => { resolve('data'); });
      });
    };

    // Both calls start before factory resolves — they share one in-flight promise
    const [a, b] = await Promise.all([
      coalesce.run('key', factory),
      coalesce.run('key', factory)
    ]);

    console.log('Coalesce shared result a:', a, 'b:', b, 'callCount:', callCount);
    assert.equal(a, 'data');
    assert.equal(b, 'data');
    assert.equal(callCount, 1);
  }

  static async runIsInflight(): Promise<void> {
    const coalesce = Coalesce.create<number>();

    let resolve!: (v: number) => void;
    const factory = (): Promise<number> =>
    {return new Promise<number>((res) => { resolve = res; });};

    // Not in-flight before we start
    const beforeStart = coalesce.isInflight('item');
    const promise = coalesce.run('item', factory);

    // In-flight while factory is pending
    const duringInflight = coalesce.isInflight('item');

    resolve(42);
    const result = await promise;

    // No longer in-flight once resolved
    const afterResolve = coalesce.isInflight('item');
    console.log('isInflight — before:', beforeStart, 'during:', duringInflight, 'after:', afterResolve, 'result:', result);
    assert.equal(result, 42);
  }

  static async runDistinctKeys(): Promise<void> {
    const coalesce = Coalesce.create<string>();
    const callCounts: Record<string, number> = { 'a': 0, 'b': 0 };

    const factory = (key: string) => {return (): Promise<string> => {
      callCounts[key]! += 1;
      return Promise.resolve(`result-${key}`);
    };};

    const [a, b] = await Promise.all([
      coalesce.run('a', factory('a')),
      coalesce.run('b', factory('b'))
    ]);

    console.log('Distinct keys — a:', a, 'b:', b, 'callCounts:', callCounts);
    assert.equal(a, 'result-a');
    assert.equal(b, 'result-b');
    assert.equal(callCounts.a, 1);
    assert.equal(callCounts.b, 1);
  }

  static async runSequentialCallsEachInvokeFactory(): Promise<void> {
    const coalesce = Coalesce.create<number>();
    let callCount = 0;

    const factory = (): Promise<number> => {
      callCount += 1;
      return Promise.resolve(callCount);
    };

    // Sequential calls — no in-flight overlap — each invokes factory
    const first = await coalesce.run('seq', factory);
    const second = await coalesce.run('seq', factory);

    console.log('Sequential — first:', first, 'second:', second, 'callCount:', callCount);
    assert.equal(first, 1);
    assert.equal(second, 2);
    assert.equal(callCount, 2);
  }
}

Timeout: `CoalesceTimeoutError`

A timeout option caps how long an individual caller waits on the shared in-flight promise. When a caller's timeout elapses, only that caller's run() rejects with CoalesceTimeoutError — the in-flight entry is left untouched for other callers still waiting on it:

typescript

import { Coalesce, CoalesceTimeoutError } from '@studnicky/concurrency';

const coalesce = Coalesce.create<Response>({ timeout: 5000 });

try {
  await coalesce.run('user:42', () => fetch('/api/user/42'));
} catch (error) {
  if (error instanceof CoalesceTimeoutError) {
    console.log(error.key);        // 'user:42'
    console.log(error.timeoutMs);  // 5000
  }
}

AsyncIter: merge, filter, enrich

Compose async iterables with FIFO merge, sync/async predicate filter, and left-join enrichment:

import { AsyncIter } from '../src/index.js';

// Native async sources — `async function*` is an AsyncIterable in browsers and
// Node alike, so AsyncIter's combinators consume them directly (no node:stream).
// Each awaits a microtask before yielding to model a genuinely asynchronous
// producer.
class AsyncSources {
  static async *range(start: number, end: number): AsyncGenerator<number> {
    for (let i = start; i <= end; i += 1) {
      await Promise.resolve();
      yield i;
    }
  }

  static async *words(items: string[]): AsyncGenerator<string> {
    for (const item of items) {
      await Promise.resolve();
      yield item;
    }
  }

  static async *itemsOf<T>(items: T[]): AsyncGenerator<T> {
    for (const item of items) {
      await Promise.resolve();
      yield item;
    }
  }
}

class AsyncIterDemo {
  static async runMerge(): Promise<void> {
    // Merge two non-overlapping ranges
    const merged = AsyncIter.merge(
      AsyncSources.range(1, 3),
      AsyncSources.range(10, 12)
    );
    const results: number[] = [];
    for await (const n of merged) {
      results.push(n);
    }

    // All 6 values present — order is arrival-based (FIFO within each source), so check membership
    assert.equal(results.length, 6);
    const expected = [1, 2, 3, 10, 11, 12];
    const expectedLen = expected.length;
    const resultsSet = new Set(results);

    for (let i = 0; i < expectedLen; i += 1) {
      assert.ok(resultsSet.has(expected[i]!), `Missing value ${expected[i]}`);
    }
    // Within each source, relative order is preserved
    assert.ok(results.indexOf(1) < results.indexOf(2));
    assert.ok(results.indexOf(2) < results.indexOf(3));
    assert.ok(results.indexOf(10) < results.indexOf(11));
    assert.ok(results.indexOf(11) < results.indexOf(12));

    console.log('merge results:', results);
  }

  static async runMergeEmpty(): Promise<void> {
    const results: number[] = [];
    for await (const n of AsyncIter.merge<number>()) {
      results.push(n);
    }
    assert.equal(results.length, 0);
    console.log('merge empty: []');
  }

  static async runFilter(): Promise<void> {
    const evens = AsyncIter.filter(AsyncSources.range(1, 10), (n) => { return n % 2 === 0; });
    const results: number[] = [];
    for await (const n of evens) {
      results.push(n);
    }
    assert.deepEqual(results, [2, 4, 6, 8, 10]);
    console.log('filter evens:', results);
  }

  static async runFilterAsync(): Promise<void> {
    // Predicate may return a Promise
    const longWords = AsyncIter.filter(
      AsyncSources.words(['hi', 'hello', 'hey', 'greetings']),
      async (w) => { const result = await Promise.resolve(w.length > 3); return result; }
    );
    const results: string[] = [];
    for await (const w of longWords) {
      results.push(w);
    }
    assert.deepEqual(results, ['hello', 'greetings']);
    console.log('filter long words:', results);
  }

  static async runEnrich(): Promise<void> {
    const items = AsyncSources.itemsOf<ItemEntity.Type>([{ 'id': 1 }, { 'id': 2 }, { 'id': 3 }]);

    // Only even ids get enrichment; odd ids pass through unchanged
    const enriched = AsyncIter.enrich<ItemEntity.Type, EnrichmentEntity.Type, EnrichedEntity.Type>(
      items,
      (item) => { const result = Promise.resolve(item.id % 2 === 0 ? { 'label': `even-${item.id}` } : null); return result; },
      (item, enrichment) => { return { 'id': item.id, 'label': enrichment.label }; }
    );

    const results: (ItemEntity.Type | EnrichedEntity.Type)[] = [];
    for await (const item of enriched) {
      results.push(item);
    }

    assert.equal(results.length, 3);

    // id:1 — unenriched, passes through as-is
    assert.deepEqual(results[0], { 'id': 1 });

    // id:2 — enriched
    assert.deepEqual(results[1], { 'id': 2, 'label': 'even-2' });

    // id:3 — unenriched
    assert.deepEqual(results[2], { 'id': 3 });

    console.log('enrich results:', results);
  }

  static async runMergeFilterEnrichPipeline(): Promise<void> {
    // Compose all three combinators in a pipeline
    const merged = AsyncIter.merge<number>(
      AsyncSources.range(1, 5),
      AsyncSources.range(6, 10)
    );
    const filtered = AsyncIter.filter<number>(merged, (n) => { return n % 3 === 0; });
    const enriched = AsyncIter.enrich<number, { 'tier': string }, { 'n': number; 'tier': string }>(
      filtered,
      (n) => { const result = Promise.resolve(n > 5 ? { 'tier': 'high' as const } : null); return result; },
      (n, e) => { return { 'n': n, 'tier': e.tier }; }
    );

    const results: ({ 'n': number; 'tier': string } | number)[] = [];
    for await (const item of enriched) {
      results.push(item);
    }

    // Multiples of 3 in 1..10: 3, 6, 9
    // 3 → unenriched (≤5)
    // 6 → enriched { n:6, tier:'high' }
    // 9 → enriched { n:9, tier:'high' }
    // merge() yields in arrival order across concurrent sources — sort for stable assertions
    const sorted = [...results].sort((a, b) => {
      const nA = typeof a === 'number' ? a : a.n;
      const nB = typeof b === 'number' ? b : b.n;
      return nA - nB;
    });
    assert.equal(sorted.length, 3);
    assert.deepEqual(sorted[0], 3);
    assert.deepEqual(sorted[1], { 'n': 6, 'tier': 'high' });
    assert.deepEqual(sorted[2], { 'n': 9, 'tier': 'high' });

    console.log('merge+filter+enrich pipeline:', sorted);
  }
}

Observability hooks

Each class exposes protected hook methods you can override in a subclass to observe internal lifecycle events without modifying the class logic.

Semaphore hooks

Hook	When it fires	Args
`onAcquire`	Permit granted immediately	`permitsBefore: number`
`onAcquireWait`	Caller queued (no permit available)	—
`onContended`	New waiter added to queue	`queueLength: number`
`onRelease`	Permit returned to pool (no waiting callers)	`permitsAfter: number`
`onReleaseDelegated`	Permit handed to queued waiter	—

Channel hooks

Hook	When it fires	Args
`onEnqueue`	Item added to buffer	`key: string, item: T`
`onDequeue`	Item removed from buffer by subscriber	`key: string, item: T`
`onPublishDropped`	Publish attempted on closed channel	`key: string, item: T`
`onClose`	Channel closes (all keys)	—

Coalesce hooks

Hook	When it fires	Args
`onCoalesceStart`	Leader caller invokes the factory	`key: string`
`onCoalesceJoin`	Caller joined an in-flight call	`key: string`
`onCoalesceSettled`	In-flight promise settled	`key: string, success: boolean`

import { Channel } from '../src/Channel.js';
import { Coalesce } from '../src/Coalesce.js';
import { Semaphore } from '../src/Semaphore.js';

class ObservedSemaphore extends Semaphore {
  readonly acquireEvents: number[] = [];
  readonly acquireWaitEvents: number[] = [];
  readonly contendedEvents: number[] = [];
  readonly releaseEvents: number[] = [];
  readonly releaseDelegatedEvents: number[] = [];

  constructor(permits: number) { super({ 'permits': permits }); }

  protected override onAcquire(permitsBefore: number): void {
    this.acquireEvents.push(permitsBefore);
    console.log(`[concurrency:semaphore] acquire permitsBefore=${permitsBefore}`);
  }
  protected override onAcquireWait(): void {
    this.acquireWaitEvents.push(1);
    console.log('[concurrency:semaphore] acquireWait');
  }
  protected override onContended(queueLength: number): void {
    this.contendedEvents.push(queueLength);
    console.log(`[concurrency:semaphore] contended queueLength=${queueLength}`);
  }
  protected override onRelease(permitsAfter: number): void {
    this.releaseEvents.push(permitsAfter);
    console.log(`[concurrency:semaphore] release permitsAfter=${permitsAfter}`);
  }
  protected override onReleaseDelegated(): void {
    this.releaseDelegatedEvents.push(1);
    console.log('[concurrency:semaphore] releaseDelegated');
  }
}

class ObservedChannel<T> extends Channel<T> {
  readonly dequeueEvents: { 'item': T; 'key': string }[] = [];
  readonly droppedEvents: { 'item': T; 'key': string }[] = [];
  readonly enqueueEvents: { 'item': T; 'key': string }[] = [];
  closeCount = 0;

  constructor() { super(); }

  protected override onDequeue(key: string, item: T): void {
    this.dequeueEvents.push({ 'item': item, 'key': key });
    console.log(`[concurrency:channel] dequeue key=${key} item=${JSON.stringify(item)}`);
  }
  protected override onEnqueue(key: string, item: T): void {
    this.enqueueEvents.push({ 'item': item, 'key': key });
    console.log(`[concurrency:channel] enqueue key=${key} item=${JSON.stringify(item)}`);
  }
  protected override onPublishDropped(key: string, item: T): void {
    this.droppedEvents.push({ 'item': item, 'key': key });
    console.log(`[concurrency:channel] publishDropped key=${key} item=${JSON.stringify(item)}`);
  }
  protected override onClose(): void {
    this.closeCount += 1;
    console.log('[concurrency:channel] close');
  }
}

class ObservedCoalesce<T> extends Coalesce<T> {
  readonly joinEvents: string[] = [];
  readonly settledEvents: { 'key': string; 'success': boolean }[] = [];
  readonly startEvents: string[] = [];

  constructor() { super(); }

  protected override onCoalesceJoin(key: string): void {
    this.joinEvents.push(key);
    console.log(`[concurrency:coalesce] join key=${key}`);
  }
  protected override onCoalesceSettled(key: string, success: boolean): void {
    this.settledEvents.push({ 'key': key, 'success': success });
    console.log(`[concurrency:coalesce] settled key=${key} success=${success}`);
  }
  protected override onCoalesceStart(key: string): void {
    this.startEvents.push(key);
    console.log(`[concurrency:coalesce] start key=${key}`);
  }
}

class ObservedConcurrencyDemo {
  static async runSemaphore(): Promise<ObservedSemaphore> {
    console.log('\n=== Semaphore ===');
    const sem = new ObservedSemaphore(1);

    // First acquire: immediate grant
    const r1 = await sem.acquire();

    // Second acquire: no permits, must wait
    const pendingR2 = sem.acquire();
    // Give microtasks a chance to run the queued waiter registration
    await Promise.resolve();

    // Release first permit: delegates to queued waiter
    await r1();

    // Wait for the second acquire to resolve
    const r2 = await pendingR2;

    // Release the second permit: returns to pool
    await r2();
    return sem;
  }

  static async runChannel(): Promise<ObservedChannel<string>> {
    console.log('\n=== Channel ===');
    const ch: ObservedChannel<string> = new ObservedChannel<string>();
    await ch.publish('events', 'hello');
    await ch.publish('events', 'world');

    const collected: string[] = [];
    const gen: AsyncGenerator<string> = ch.subscribe('events');
    for await (const item of gen) {
      collected.push(item);
      if (collected.length >= 2) { break; }
    }

    await ch.close();
    console.log('Channel received:', collected);
    return ch;
  }

  static async runCoalesce(): Promise<ObservedCoalesce<string>> {
    console.log('\n=== Coalesce ===');
    const coalesce: ObservedCoalesce<string> = new ObservedCoalesce<string>();

    // 3 concurrent calls for same key: 1 leader + 2 joiners
    const factory = (): Promise<string> => {
      return new Promise<string>((resolve) => {
        setImmediate(() => { resolve('v'); });
      });
    };

    await Promise.all([
      coalesce.run('batch', factory),
      coalesce.run('batch', factory),
      coalesce.run('batch', factory)
    ]);

    // Sequential call: new leader
    await coalesce.run('batch', factory);
    return coalesce;
  }
}

The base class never calls any logger or metrics library. All hooks are no-ops by default.

Try it

The builder demo constructs a Semaphore via Semaphore.builder().withPermits(2).build() and a Channel via Channel.builder().build(). Watch the Semaphore limit concurrent executions to two at a time across four competing tasks, then watch the Channel deliver three buffered items in publish order.

Semaphore and Channel builders

/** builderConcurrency — constructs a Semaphore via Semaphore.builder() and a Channel via Channel.builder(), both exercised with visible output. Run: npx tsx examples/builderConcurrency.ts */

import assert from 'node:assert/strict';

// #region usage
import { Channel, Semaphore } from '../src/index.js';

class Task {
  static readonly completions: number[] = [];
  static maxConcurrent = 0;
  static #inFlight = 0;

  static async run(id: number, sem: Semaphore): Promise<void> {
    const result = sem.withPermit(async () => {
      Task.#inFlight += 1;
      Task.maxConcurrent = Math.max(Task.maxConcurrent, Task.#inFlight);
      // Yield to the microtask queue so tasks interleave
      await Promise.resolve();
      Task.#inFlight -= 1;
      Task.completions.push(id);
    });
    return await result;
  }
}

class BuilderConcurrencyDemo {
  static async runSemaphore(): Promise<void> {
    // --- Semaphore via builder ---
    const sem = Semaphore.builder()
      .withPermits(2)
      .build();

    console.log('Semaphore built. Permits:', sem.permits, 'Available:', sem.available);

    // 4 concurrent tasks against a 2-permit semaphore — max concurrent stays at or below 2
    await Promise.all([Task.run(1, sem), Task.run(2, sem), Task.run(3, sem), Task.run(4, sem)]);
    console.log('Max concurrent (≤2):', Task.maxConcurrent);
    console.log('Completions:', Task.completions);

    assert.ok(Task.maxConcurrent <= 2, `maxConcurrent ${Task.maxConcurrent} exceeded permits`);
    assert.equal(Task.completions.length, 4, 'All 4 tasks completed');
  }

  static async runChannel(): Promise<void> {
    // --- Channel via builder ---
    const channel = Channel.builder<string>().build();

    console.log('\nChannel built. Publishing items...');
    await channel.publish('notifications', 'signup');
    await channel.publish('notifications', 'payment');
    await channel.publish('notifications', 'logout');
    await channel.close();

    const received: string[] = [];
    for await (const item of channel.subscribe('notifications')) {
      received.push(item);
    }
    console.log('Channel received:', received);
    assert.deepEqual(received, ['signup', 'payment', 'logout'], 'Channel delivered all items in order');
  }
}
// #endregion usage

await BuilderConcurrencyDemo.runSemaphore();
await BuilderConcurrencyDemo.runChannel();

console.log('builderConcurrency: all assertions passed');

Output

Press Execute to run this example against the real library.

The async-iter demo uses native async function* generators as sources — no Node.js streams — and passes them through AsyncIter.merge, AsyncIter.filter, and AsyncIter.enrich. Watch the merged output interleave values from two independent ranges, the filter keep only even numbers, and the final composed pipeline emit only the multiples-of-three with a tier enrichment applied to values above five.

AsyncIter merge / filter / enrich

/** asyncIter — demonstrates AsyncIter.merge, AsyncIter.filter, and AsyncIter.enrich. Run: npx tsx examples/asyncIter.ts */

import assert from 'node:assert/strict';

import type { EnrichedEntity } from './entities/EnrichedEntity.js';
import type { EnrichmentEntity } from './entities/EnrichmentEntity.js';
import type { ItemEntity } from './entities/ItemEntity.js';

// #region usage
import { AsyncIter } from '../src/index.js';

// Native async sources — `async function*` is an AsyncIterable in browsers and
// Node alike, so AsyncIter's combinators consume them directly (no node:stream).
// Each awaits a microtask before yielding to model a genuinely asynchronous
// producer.
class AsyncSources {
  static async *range(start: number, end: number): AsyncGenerator<number> {
    for (let i = start; i <= end; i += 1) {
      await Promise.resolve();
      yield i;
    }
  }

  static async *words(items: string[]): AsyncGenerator<string> {
    for (const item of items) {
      await Promise.resolve();
      yield item;
    }
  }

  static async *itemsOf<T>(items: T[]): AsyncGenerator<T> {
    for (const item of items) {
      await Promise.resolve();
      yield item;
    }
  }
}

class AsyncIterDemo {
  static async runMerge(): Promise<void> {
    // Merge two non-overlapping ranges
    const merged = AsyncIter.merge(
      AsyncSources.range(1, 3),
      AsyncSources.range(10, 12)
    );
    const results: number[] = [];
    for await (const n of merged) {
      results.push(n);
    }

    // All 6 values present — order is arrival-based (FIFO within each source), so check membership
    assert.equal(results.length, 6);
    const expected = [1, 2, 3, 10, 11, 12];
    const expectedLen = expected.length;
    const resultsSet = new Set(results);

    for (let i = 0; i < expectedLen; i += 1) {
      assert.ok(resultsSet.has(expected[i]!), `Missing value ${expected[i]}`);
    }
    // Within each source, relative order is preserved
    assert.ok(results.indexOf(1) < results.indexOf(2));
    assert.ok(results.indexOf(2) < results.indexOf(3));
    assert.ok(results.indexOf(10) < results.indexOf(11));
    assert.ok(results.indexOf(11) < results.indexOf(12));

    console.log('merge results:', results);
  }

  static async runMergeEmpty(): Promise<void> {
    const results: number[] = [];
    for await (const n of AsyncIter.merge<number>()) {
      results.push(n);
    }
    assert.equal(results.length, 0);
    console.log('merge empty: []');
  }

  static async runFilter(): Promise<void> {
    const evens = AsyncIter.filter(AsyncSources.range(1, 10), (n) => { return n % 2 === 0; });
    const results: number[] = [];
    for await (const n of evens) {
      results.push(n);
    }
    assert.deepEqual(results, [2, 4, 6, 8, 10]);
    console.log('filter evens:', results);
  }

  static async runFilterAsync(): Promise<void> {
    // Predicate may return a Promise
    const longWords = AsyncIter.filter(
      AsyncSources.words(['hi', 'hello', 'hey', 'greetings']),
      async (w) => { const result = await Promise.resolve(w.length > 3); return result; }
    );
    const results: string[] = [];
    for await (const w of longWords) {
      results.push(w);
    }
    assert.deepEqual(results, ['hello', 'greetings']);
    console.log('filter long words:', results);
  }

  static async runEnrich(): Promise<void> {
    const items = AsyncSources.itemsOf<ItemEntity.Type>([{ 'id': 1 }, { 'id': 2 }, { 'id': 3 }]);

    // Only even ids get enrichment; odd ids pass through unchanged
    const enriched = AsyncIter.enrich<ItemEntity.Type, EnrichmentEntity.Type, EnrichedEntity.Type>(
      items,
      (item) => { const result = Promise.resolve(item.id % 2 === 0 ? { 'label': `even-${item.id}` } : null); return result; },
      (item, enrichment) => { return { 'id': item.id, 'label': enrichment.label }; }
    );

    const results: (ItemEntity.Type | EnrichedEntity.Type)[] = [];
    for await (const item of enriched) {
      results.push(item);
    }

    assert.equal(results.length, 3);

    // id:1 — unenriched, passes through as-is
    assert.deepEqual(results[0], { 'id': 1 });

    // id:2 — enriched
    assert.deepEqual(results[1], { 'id': 2, 'label': 'even-2' });

    // id:3 — unenriched
    assert.deepEqual(results[2], { 'id': 3 });

    console.log('enrich results:', results);
  }

  static async runMergeFilterEnrichPipeline(): Promise<void> {
    // Compose all three combinators in a pipeline
    const merged = AsyncIter.merge<number>(
      AsyncSources.range(1, 5),
      AsyncSources.range(6, 10)
    );
    const filtered = AsyncIter.filter<number>(merged, (n) => { return n % 3 === 0; });
    const enriched = AsyncIter.enrich<number, { 'tier': string }, { 'n': number; 'tier': string }>(
      filtered,
      (n) => { const result = Promise.resolve(n > 5 ? { 'tier': 'high' as const } : null); return result; },
      (n, e) => { return { 'n': n, 'tier': e.tier }; }
    );

    const results: ({ 'n': number; 'tier': string } | number)[] = [];
    for await (const item of enriched) {
      results.push(item);
    }

    // Multiples of 3 in 1..10: 3, 6, 9
    // 3 → unenriched (≤5)
    // 6 → enriched { n:6, tier:'high' }
    // 9 → enriched { n:9, tier:'high' }
    // merge() yields in arrival order across concurrent sources — sort for stable assertions
    const sorted = [...results].sort((a, b) => {
      const nA = typeof a === 'number' ? a : a.n;
      const nB = typeof b === 'number' ? b : b.n;
      return nA - nB;
    });
    assert.equal(sorted.length, 3);
    assert.deepEqual(sorted[0], 3);
    assert.deepEqual(sorted[1], { 'n': 6, 'tier': 'high' });
    assert.deepEqual(sorted[2], { 'n': 9, 'tier': 'high' });

    console.log('merge+filter+enrich pipeline:', sorted);
  }
}
// #endregion usage

await AsyncIterDemo.runMerge();
await AsyncIterDemo.runMergeEmpty();
await AsyncIterDemo.runFilter();
await AsyncIterDemo.runFilterAsync();
await AsyncIterDemo.runEnrich();
await AsyncIterDemo.runMergeFilterEnrichPipeline();

console.log('asyncIter: all assertions passed');

Output

Press Execute to run this example against the real library.

API

Export	Type	Description
`Channel<T>`	class	String-keyed fan-in async generator inbox
`Semaphore`	class	Counting permit gate with async acquire
`Coalesce<T>`	class	Deduplicates concurrent calls for the same key
`AsyncIter`	class	Static utilities for async iterables

`Channel<T>`

Member	Signature	Description
`publish`	`(key: string, item: T) => void`	Sends item to subscribers on `key`
`subscribe`	`(key: string) => AsyncGenerator<T>`	Yields items published to `key`
`close`	`() => void`	Closes all channels; subscribers stop after draining

`Semaphore`

Member	Signature	Description
`acquire`	`() => Promise<() => void>`	Waits for a permit; returns a release function
`withPermit`	`<T>(callback: () => Promise<T>) => Promise<T>`	Acquires, runs callback, releases
`available`	`number`	Current available permit count
`permits`	`number`	Total configured permits

`Coalesce<T>`

Member	Signature	Description
`run`	`(key: string, factory: () => Promise<T>) => Promise<T>`	Shares in-flight promise for `key`
`isInflight`	`(key: string) => boolean`	True if a promise for `key` is pending

`AsyncIter`

Member	Signature	Description
`merge`	`<T>(...sources) => AsyncGenerator<T>`	FIFO merge of N async iterables
`filter`	`<T>(source, predicate) => AsyncGenerator<T>`	Yields items matching sync/async predicate
`enrich`	`<T, E, R>(source, lookup, merge) => AsyncGenerator<T \| R>`	Left-join enrichment per item

Source on GitHub

@studnicky/concurrency ​

Install ​

Usage ​

Channel and Semaphore ​

Coalesce: deduplicate concurrent calls by key ​

Timeout: CoalesceTimeoutError ​

AsyncIter: merge, filter, enrich ​

Observability hooks ​

Semaphore hooks ​

Channel hooks ​

Coalesce hooks ​

Try it ​

API ​

Channel<T> ​

Semaphore ​

Coalesce<T> ​

AsyncIter ​