@torquedev/ext-async-events

Async job queue enhancement for @torquedev/eventbus. Adds subscribeAsync() with retry logic so event handlers run outside the synchronous publish() call chain, deferred to the next event-loop tick via setImmediate with exponential backoff on failure.

Install

npm install github:michaeljabbour/torque-ext-async-events

Peer dependency: @torquedev/eventbus >=0.1.0 Runtime dependencies: none (zero-dep package)

Usage

Patch an existing EventBus class at boot:

import { applyAsync } from '@torquedev/ext-async-events';
import { EventBus } from '@torquedev/eventbus';

applyAsync(EventBus);

Then use subscribeAsync in bundles:

eventBus.subscribeAsync(
  'pipeline.deal.stage_changed',
  'communications',
  async (event) => {
    await sendSMS(event.data.dealId);
  },
  { maxRetries: 5 }
);

API

applyAsync(EventBusClass, options?)

Monkey-patches async capabilities onto an existing EventBus class. Adds the methods below to its prototype.

subscribeAsync(eventName, subscriberName, handler, options?)

Subscribe to an event with async execution. The handler runs after the current tick, not inline with publish().

• eventName -- event to subscribe to
• subscriberName -- identifier for this subscriber
• handler -- async (event) => void
• options.maxRetries -- override the default max retries (default: 3)

getJobStats()

Returns queue statistics: { queued, processing, completed, failed, retrying }.

getFailedJobs()

Returns the failed job log: [{ subscriberName, eventName, error, attempts, timestamp }].

drain()

Returns a Promise that resolves when all in-flight and retrying jobs complete. Use for graceful shutdown.

AsyncEventBus

Standalone event bus with both subscribe (sync) and subscribeAsync built in. Useful for testing:

import { AsyncEventBus } from '@torquedev/ext-async-events';

const bus = new AsyncEventBus();
bus.subscribeAsync('user.created', 'welcome-email', handler);
bus.publish('user.created', { userId: '123' });
await bus.drain();

InProcessQueue

The underlying queue implementation. Uses setImmediate-based deferred execution with exponential backoff retry (baseDelay * attempts^2).

import { InProcessQueue } from '@torquedev/ext-async-events';

const queue = new InProcessQueue({
  maxRetries: 3,
  baseDelay: 1000,
  onFailed: (job) => console.error(job),
});

Exports

import { applyAsync, AsyncEventBus, InProcessQueue } from '@torquedev/ext-async-events';

Testing

npm test

ESM-only. Requires Node.js with the built-in test runner.

License

MIT