@bemedev/subscriber

A feature-rich, lifecycle-aware subscription manager supporting RxJS / Subscribable interoperability, fluent selector chaining, custom equality comparators, state control, and explicit resource disposal (Disposable).

Features

• ⚡ RxJS / Observable Integration: Subscribe seamlessly to any Subscribable source (RxJS Observables, Subjects, BehaviorSubjects, or custom emitters).
• 🔗 Fluent Builder & Selector Chaining: Chain nested state selectors using .select(selector) to compute derived state and only react to specific sub-state changes.
• 🔍 Custom Equality Comparators: Prevent redundant subscriber notifications by comparing previous and current values using custom comparator functions or strict equality (normalEquals).
• ⏸ Lifecycle Control: Granular state management across lifecycle states (active, paused, inactive, disposed) using open(), close(), unsubscribe(), and dispose().
• 🧹 Explicit Resource Disposal: Native support for JavaScript/TypeScript Disposable (Symbol.dispose and Symbol.asyncDispose) for use with using declarations.

Installation

npm install @bemedev/subscriber
# or
pnpm add @bemedev/subscriber
# or
yarn add @bemedev/subscriber

Usage

Basic Usage with RxJS Subject

import { createSubscriber } from '@bemedev/subscriber';
import { Subject } from 'rxjs';

// Create a subject source
const source$ = new Subject<number>();

// Create a subscriber node from source
const subscriber = createSubscriber(source$).subscribe(val => {
  console.log(`Received value: ${val}`);
});

// Emit values from source
source$.next(1); // Logs: "Received value: 1"
source$.next(1); // Skipped due to equality check (default normalEquals)
source$.next(2); // Logs: "Received value: 2"

Selector Chaining

import { createSubscriber } from '@bemedev/subscriber';
import { BehaviorSubject } from 'rxjs';

type State = { user: { name: string; age: number } };
const state$ = new BehaviorSubject<State>({
  user: { name: 'Alice', age: 30 },
});

// Chain selectors to transform emission values
const subscriber = createSubscriber(state$)
  .select(state => state.user)
  .select(user => user.name)
  .subscribe(name => console.log(`Name: ${name}`));

state$.next({ user: { name: 'Alice', age: 31 } }); // Skipped (selected name unchanged)
state$.next({ user: { name: 'Bob', age: 31 } }); // Logs: "Name: Bob"

Custom Equality Comparator

import { createSubscriber } from '@bemedev/subscriber';
import { Subject } from 'rxjs';

type User = { id: string; name: string };
const source$ = new Subject<User>();

// Ignore notifications if user ID hasn't changed
const subscriber = createSubscriber(source$).subscribe(
  user => console.log(`User updated: ${user.name}`),
  (prev, curr) => prev?.id === curr?.id,
);

source$.next({ id: '1', name: 'Alice' }); // Logs: "User updated: Alice"
source$.next({ id: '1', name: 'Alice Smith' }); // Skipped (same ID)

Lifecycle Control

import { createSubscriber } from '@bemedev/subscriber';

const subscriber = createSubscriber(source$).subscribe(val =>
  console.log(val),
);

// Pause notifications
subscriber.close(); // state becomes 'paused'

// Resume notifications
subscriber.open(); // state becomes 'active'

// Deactivate subscriber
subscriber.unsubscribe(); // state becomes 'inactive'

// Re-subscribe when inactive
subscriber.reSubscribe(); // state becomes 'active'

// Permanently dispose subscriber resources
subscriber.dispose(); // state becomes 'disposed'

Explicit Resource Management (using)

import { createSubscriber } from '@bemedev/subscriber';

function run() {
  using subscriber = createSubscriber(source$).subscribe(val =>
    console.log(val),
  );
  // subscriber automatically disposes when scope exits via Symbol.dispose
}

API Reference

createSubscriber(subscribable)

Factory function to create a new SubscriberBuilder instance attached to a subscribable source.

• subscribable: Subscribable<T> — Source subscribable object.
• Returns: SubscriberBuilder<T, T>

createManagedSubscriber(subscriber, options?)

Factory function to create a new ManagedSubscriber instance.

• subscriber: Subscriber_F<T> — Subscriber callback function.
• options: SusbscriberOptions<T, R> — Optional configuration (equality comparator, selector).
• Returns: ManagedSubscriber<T, R>

defaultSelector(a)

Identity selector function that returns the input value unchanged.

• a: T — Input value.
• Returns: R — Input value cast to output type R.

normalEquals(a, b)

Strict equality (===) comparator function used as default equality checker.

• a: T — First value to compare.
• b: T — Second value to compare.
• Returns: boolean — true if a === b, otherwise false.

SubscriberBuilder<T, R>

Builder class used to chain selector transformations and create active subscribers.

| Method / Property | Type / Return | Description | | -------------------------------- | ----------------------------- | -------------------------------------------------------------------------------------- | | subscribable | Subscribable<T> | Returns attached source subscribable object | | select(selector) | SubscriberBuilder<T, RNext> | Creates a new subscriber builder with a nested selector transformer | | subscribe(subscriber, equals?) | Subscriber<T, R> | Subscribes callback to state updates with optional equality comparator and starts node |

Subscriber<T, R>

Main class representing an active subscriber node (extends BaseSubscriber).

| Method / Property | Type / Return | Description | | ------------------------- | ------------------------------- | -------------------------------------------------------------------------- | | state | SubscriberState | Returns current state ('active', 'paused', 'inactive', 'disposed') | | equals | Equals_F<R> | Returns equality comparator function | | selector | Selector_F<T, R> \| undefined | Returns selector function or undefined | | subscribable | Subscribable<T> \| undefined | Returns source subscribable object or undefined if disposed | | isNotInactive | boolean | true if state is neither 'disposed' nor 'inactive' | | close() | SubscriberState | Pauses subscriber notifications ('paused') | | open() | SubscriberState | Resumes subscriber notifications ('active') | | unsubscribe() | SubscriberState | Unsubscribes subscriber ('inactive') | | reSubscribe() | SubscriberState | Re-subscribes to source subscribable if inactive ('active') | | dispose() | SubscriberState | Cleans up subscriber references and sets state to 'disposed' | | [Symbol.dispose]() | SubscriberState | Standard synchronous disposal | | [Symbol.asyncDispose]() | Promise<SubscriberState> | Standard asynchronous disposal |

BaseSubscriber<T, R> (@bemedev/subscriber/base)

Abstract base class for subscribers managing subscription state, equality checking, and disposal.

ManagedSubscriber<T, R> (@bemedev/subscriber/managed)

Managed subscriber class providing subscription lifecycle control methods (close(), open(), unsubscribe(), dispose()).

License

MIT

CHANGELOG

Read CHANGELOG.md for more details about the changes.

Author

chlbri ([email protected])

My GitHub

Links

• Documentation