@pawells/rxjs-events

RxJS Events

Description

RxJS-based event handling library with reactive observables and async event streams. Provides a typed EventHandler class for publish/subscribe event management, AsyncObservable for backpressure-aware async iteration, DebounceEvents/ThrottleEvents/PipeEvents for composing and transforming handler output, and a suite of async-iterator-based utilities for chunking, partitioning, and grouping event streams.

The package also exports a ./mocks entry point containing Vitest custom matchers and mock factories for use in test files.

The package is ESM-only, targets Node.js >= 22, and ships with full TypeScript declarations.

Requirements

• Node.js: >= 22.x
• TypeScript: >= 6.x
• Module system: ESM ("type": "module" — CommonJS is not supported)
• Dependencies: rxjs >= 7.8.2 and @pawells/typescript-common are regular dependencies (not peer dependencies) — both are installed automatically when you install this package.

Installation

npm install @pawells/rxjs-events
# or
yarn add @pawells/rxjs-events

Quick Start

Basic event handler

import { EventHandler, type TEventData } from '@pawells/rxjs-events';

interface UserData {
  id: number;
  name: string;
}

interface UserEvent extends TEventData {
  UserCreated: UserData;
}

const handler = new EventHandler<UserData, UserEvent>('UserCreated');

// Subscribe to events
const handle = handler.Subscribe(async (event) => {
  console.log('User created:', event.UserCreated.name);
});

// Trigger an event
handler.Trigger({ id: 1, name: 'Alice' });

// Unsubscribe when done
handler.Unsubscribe(handle);

Async iteration

import { EventHandler, type TEventData } from '@pawells/rxjs-events';

interface TickEvent extends TEventData {
  Tick: { value: number };
}

const handler = new EventHandler<{ value: number }, TickEvent>('Tick');

// Consume events via async iterator — errors propagate in-band
(async () => {
  for await (const event of handler.GetAsyncIterableIterator()) {
    console.log('Tick:', event.Tick.value);
    if (event.Tick.value >= 5) break;
  }
})();

for (let i = 1; i <= 5; i++) {
  handler.Trigger({ value: i });
}

Filtering events

import {
  EventHandler,
  PartitionEvents,
  type TEventData,
} from '@pawells/rxjs-events';

interface MessageEvent extends TEventData {
  MessageReceived: { channel: string; text: string };
}

const handler = new EventHandler<
  { channel: string; text: string },
  MessageEvent
>('MessageReceived');

// Split the stream into two independently-consumable async iterators based on a predicate
const [general, other] = PartitionEvents(
  handler,
  (event) => event.MessageReceived.channel === 'general',
);

(async () => {
  for await (const event of general) {
    console.log('General:', event.MessageReceived.text);
  }
})();

handler.Trigger({ channel: 'general', text: 'Hello' });

Test matchers

import { MockEventHandler } from '@pawells/rxjs-events/mocks';
import type { TEventData } from '@pawells/rxjs-events';

interface UpdatedEvent extends TEventData {
  Updated: { value: number };
}

// In your Vitest setup file — the /mocks entry point registers the custom
// matchers (toHaveSubscribers, toHaveTriggeredEvent, toMatchEventFilter) as
// a side effect of import, so a plain side-effect import is enough:
// import '@pawells/rxjs-events/mocks';

// In tests:
const mock = new MockEventHandler<{ value: number }, UpdatedEvent>('Updated');
mock.Subscribe(() => {});
mock.Trigger({ value: 42 });

expect(mock).toHaveTriggeredEvent('Updated', { value: 42 });
expect(mock).toHaveSubscribers(1);

API Reference

All symbols are exported from @pawells/rxjs-events unless noted as @pawells/rxjs-events/mocks.

---

Core — EventHandler

The primary class for typed publish/subscribe event management.

Symbol	Kind	Description
EventHandler<TObject, TEvent>	class	Wraps an RxJS Subject to provide typed event publish/subscribe. TObject is the data shape triggered; TEvent wraps it as a TEventData object.
EventHandler.Subscribe(fn, onError?)	method	Registers an event handler callback. Returns a SubscriptionHandle. onError is invoked if the underlying Subject errors; if omitted, such errors are silently swallowed. Use GetAsyncIterableIterator() for in-band propagation of errors thrown from fn itself.
EventHandler.Unsubscribe(handle)	method	Removes a subscription by its SubscriptionHandle.
EventHandler.Trigger(data)	method	Publishes data to all current subscribers.
EventHandler.GetAsyncIterableIterator()	method	Returns an async iterable that yields events one at a time. Errors propagate in-band (thrown from await). Use break or return to stop iteration. Takes no parameters.
EventHandler.GetAsyncIterator()	method	Lower-level variant of GetAsyncIterableIterator() for manual next() advancement.
EventHandler[Symbol.asyncIterator]()	method	Makes the handler itself directly usable in for await (const event of handler); delegates to GetAsyncIterableIterator().
EventHandler.Destroy()	method	Completes the internal Subject and unsubscribes all active subscriptions, releasing resources.
EventHandler.GetSubscriptionCount()	method	Returns the current number of active subscriptions.
EventHandler.GetActiveSubscriptionIds()	method	Returns a snapshot array of the SubscriptionHandles currently active.

---

Core — AsyncObservable

Backpressure-aware async iteration on top of RxJS Observable.

Symbol	Kind	Description
AsyncObservable<T>	class	Extends RxJS Observable<T>. Configured via its constructor (new AsyncObservable<T>(config?)); values are pushed with Push(value) and consumed via for await ([Symbol.asyncIterator]()).
IAsyncObservableConfig	interface	Configuration for AsyncObservable: maxBufferSize (default: 1000), overflowStrategy (a BackpressureStrategy enum member; default: BackpressureStrategy.DropOldest).
BackpressureStrategy	enum	DropOldest, DropNewest, Error — overflow behavior shared by AsyncObservable and the event-operators utilities below.
BufferOverflowError	class	Thrown when a bounded buffer overflows and the configured strategy is BackpressureStrategy.Error. Extends BaseError.
AsyncObservableError	class	Thrown when an operation (e.g. Push()) is attempted on an AsyncObservable after Destroy() has been called. Extends BaseError.

---

Core — EventFilter

Functions for matching and partitioning individual events by filter criteria.

Symbol	Kind	Description
EventFilter<TEvent>(event, criteria)	function	Returns true if the single-key event object matches every key/value pair in criteria (supports dot-notation nested paths and predicate functions), false otherwise. criteria of null/undefined always matches.
PartitionEventFilter<TEvent>(event, criteria)	function	Applies EventFilter to a single already-received event and returns a tuple [matchingEventOrNull, nonMatchingEventOrNull].
IFilterCriteria	interface	{ [key: string]: unknown } — plain key/value pairs (or predicate functions) used for event matching.
EventFilterError	class	Thrown by EventFilter/PartitionEventFilter when the event object doesn't have exactly one top-level key with a defined payload. Extends BaseError.

---

Core — Event Pipeline

Functions for composing and transforming an EventHandler's output.

Symbol	Kind	Description
DebounceEvents<TObject, TEvent>(handler, ms)	function	Returns a new EventHandler that re-triggers events from handler, delayed until ms milliseconds have passed without a new source event.
ThrottleEvents<TObject, TEvent>(handler, ms)	function	Returns a new EventHandler that re-triggers events from handler at most once per ms milliseconds.
PipeEvents<TEvent, ...>(handler, ...fns)	function	Returns an AsyncIterableIterator<TResult> that applies fns in sequence to each event from handler. Up to 3 transform functions are fully type-checked via overloads; 4+ fall back to an untyped variadic overload.

---

Core — Event Operators

Async-iterator-based utilities for common event stream transformations. Each takes an EventHandler as its first argument.

Symbol	Kind	Description
ChunkEvents<TEvent>(handler, size, backpressureConfig?)	function	Returns an AsyncIterableIterator<TEvent[]> that yields events batched into arrays of size (the final batch may be smaller).
PartitionEvents<TEvent>(handler, predicate, backpressureConfig?)	function	Returns a tuple [AsyncIterableIterator<TEvent>, AsyncIterableIterator<TEvent>] splitting handler's stream into matching/non-matching events per predicate.
GroupEventsByPayload<TEvent, TKey>(handler, keyFn, flushSize, backpressureConfig?)	function	Returns an AsyncIterableIterator<Map<TKey, TEvent[]>> grouping events by keyFn's result, yielding each group once it reaches flushSize (plus a final flush of any remainder).
IBackpressureConfig	interface	Configuration shared by the operators above: maxBufferSize?: number, overflowStrategy?: BackpressureStrategy.

---

Core — Async Iterator Helper

Low-level utilities underpinning EventHandler.GetAsyncIterableIterator() and AsyncObservable.

Symbol	Kind	Description
createAsyncIterator<T>(sourceSubscribe, config?)	function	Creates an AsyncIterableIterator<T> backed by an internal queue, subscribing lazily to sourceSubscribe on the first next() call. config is an IAsyncIteratorQueueConfig.
IAsyncIteratorQueueConfig	interface	Options: maxBufferSize?: number, overflowStrategy?: BackpressureStrategy.
IAsyncIteratorQueue<T>	interface	Internal queue-state shape (buffer, head, deferreds, hasError, error, isComplete) used by createAsyncIterator's implementation — not a consumer-facing push/pull API.
IAsyncGeneratorESN<T, TReturn, TNext>	interface	Extends AsyncGenerator with explicit-resource-management support ([Symbol.asyncDispose]()) for use with await using.

---

Types

Symbol	Kind	Description
TEventData	type	Record<string, unknown> — base constraint for event shape types.
TEventFunction<TEvent>	type	(_data: TEvent) => Promise<void> | void
TEventHandler<TPayload>	type	(payload: TPayload) => void | Promise<void>
TExtractEventPayload<TEvent>	type	Extracts the payload type from a single-key TEventData-constrained type.
SubscriptionHandle	type	Branded number identifying a subscription; returned by EventHandler.Subscribe().

---

Mocks (@pawells/rxjs-events/mocks)

Test utilities and Vitest custom matchers. Import from the separate entry point.

import {
  MockEventHandler,
  MockAsyncObservable,
} from '@pawells/rxjs-events/mocks';

Importing @pawells/rxjs-events/mocks (or any of its symbols) registers the custom matchers on Vitest's expect as a side effect of the import — there is no separate setup function to call:

// In a Vitest setup file:
import '@pawells/rxjs-events/mocks'; // registers toHaveSubscribers, toHaveTriggeredEvent, toMatchEventFilter

Symbol	Kind	Description
MockEventHandler<TObject, TEvent>	class	Test double for EventHandler. Constructor requires a non-empty name. Records all triggers and subscriptions; exposes assertion helpers (GetTriggeredEvents(), GetSubscriberCount(), GetSubscriptionCalls()).
MockEventHandlerError	class	Thrown by MockEventHandler on an empty name or exceeded maxSubscribers. Extends BaseError.
MockAsyncObservable<T>	class	Test double for AsyncObservable. Allows manual/controllable value emission (delay, error, timeout, cancellation) in tests.
MockAsyncObservableError	class	Thrown internally by MockAsyncObservable when a wait is cancelled or times out. Extends BaseError.
ToHaveSubscribers(handler, count)	function	Vitest matcher implementation — asserts that handler (a MockEventHandler) has exactly count active subscribers.
ToHaveTriggeredEvent(handler, eventType, data?)	function	Vitest matcher implementation — asserts that handler triggered an event named eventType, optionally matching data.
ToMatchEventFilter(event, criteria)	function	Vitest matcher implementation — asserts that a single-key event object matches criteria (delegates to the same filtering logic as EventFilter).
GenerateEventData<T>(eventType, countOrConfig?, config?)	function	Generates an array of typed test event data objects keyed by eventType.
GenerateFilterCriteria(complexity?)	function	Generates sample filter criteria objects for tests ('simple', 'moderate', 'complex'; default 'simple').
GenerateUserEvents(count?)	function	Generates sample userCreated event payloads.
GenerateMessageEvents(count?)	function	Generates sample messageReceived event payloads.
GenerateSubscriptionScenarios(scenarioCount?, config?)	function	Generates scenarioCount (default 5) subscription test scenarios; pass a numeric seed in config for deterministic output.
GenerateCustomFields(customFields)	function	Generates a record from a map of field-name-to-factory-function entries.
IMockEventHandlerConfig	interface	Configuration for MockEventHandler: asyncMode?, delay?, maxSubscribers?, trackCalls?.
IMockSubscription	interface	Describes a recorded subscription: id, handler, createdAt, active.
IEventDataGeneratorConfig	interface	Options for GenerateEventData: count?, seed?, customFields?.
IMockObservableConfig<T>	interface	Options for MockAsyncObservable: data, emissionDelay?, autoComplete?, shouldError?, error?, timeout?, abortSignal?.
IMockConfig	interface	General mock configuration options: maxEvents?, minDelay?, maxDelay?, includeTimestamps?.
IMockEventData	interface	Shape for generated mock event data: type, payload, timestamp?.
ICustomMatchers<R>	interface	Type declarations for the three custom Vitest matchers.

Mock Data Constants (@pawells/rxjs-events/mocks)

Symbol	Kind	Description
MOCK_USER_NAMES	constant	Pool of sample user names used by GenerateUserEvents.
MOCK_MESSAGE_CONTENT	constant	Pool of sample message text used by GenerateMessageEvents.
MOCK_EVENT_TYPES	constant	Pool of sample event type name strings.
MOCK_STATUS_VALUES	constant	Pool of sample status strings (active, inactive, pending, completed, failed).
MOCK_DOMAINS	constant	Pool of sample email domains used by GenerateUserEvents.
MOCK_ROLES	constant	Pool of sample user roles (user, admin, moderator, guest).
MOCK_PRIORITIES	constant	Pool of sample message priorities (low, normal, high).
MOCK_CHANNELS	constant	Pool of sample channel names.
DEFAULT_MOCK_CONFIG	constant	Default IMockConfig values used by GenerateSubscriptionScenarios.

License

MIT — See LICENSE for details.