Transferum

A reactive data processing pipeline system for TypeScript.

The library provides type-safe primitives for building data flows: transfers (flow nodes), bridges (connectors between flows), builders (chain constructors), and operators (data transformers).

Table of Contents

• Why Transferum?
  • Problems It Solves
  • Key Benefits
  • Use Cases
• Domain-Specific Applications
  • Game Development
  • IoT & Automation
  • UI/UX Applications
  • Monitoring & Logging Systems
  • Financial Applications
• Comparison with Alternatives
  • Transferum vs RxJS
  • Transferum vs Most.js
  • Transferum vs Bacon.js / Kefir
  • Quick Comparison Table
  • When to Choose Transferum
  • When to Consider Alternatives
• Installation & Import
• Core Concepts
  • Capability Flags System
  • Transfers
  • Linking Transfers
  • Undefined Behavior in Data Flows
  • Error Handling
• Transfers
  • Transfer Comparison Table
  • PushChannelTransfer
  • DelayedPushChannelTransfer
  • DebounceTransfer
  • ThrottleTransfer
  • PushStoredChannelTransfer
  • BufferTransfer
  • ManualBufferTransfer
  • ManualFlowTransfer
  • GateTransfer
  • MergeTransfer
  • SplitTransfer
  • PollingSourceTransfer
  • PollingProxyTransfer
  • PollingFlowTransfer
  • IdlePollingTransfer
  • ChannelTransfer
  • StoredChannelTransfer
  • SinkTransfer
  • WriteTransfer
  • ReadTransfer
  • ConvertTransfer
  • ConditionTransfer
  • UniversalCompositeTransfer
  • Async Transfer Comparison Table
  • Async Transfers
    • AsyncSinkTransfer
    • AsyncWriteTransfer
    • AsyncReadTransfer
    • AsyncConvertTransfer
    • AsyncConditionTransfer
    • AsyncPollingSourceTransfer
    • AsyncPollingProxyTransfer
    • AsyncPollingFlowTransfer
    • AsyncIdlePollingTransfer
    • AsyncStoredChannelTransfer
• Operators
  • Operator Comparison Table
  • Async Operators
• Storages
  • Storage Comparison Table
• Tickers
• Helpers
  • Subscriber
  • SubscriptionManager
  • ProxyReference
  • DisposableSubscriberAdapter
  • StateSubscriptionManager
• Bridges
  • Bridge Comparison Table
  • AsyncTransformBridge
• Pipeline Builders
  • Async Builders
• Factories
• Utilities
• Types
• Configurations
• Running Tests
• License

Why Transferum?

Problems It Solves

As data flows grow in complexity, connecting heterogeneous sources becomes a major bottleneck. Mixing push-based streams, pull-based APIs, polling loops, and async operations demands endless bespoke glue code, and this integration cost compounds with every new stage.

Transferum addresses the structural issues that emerge at scale:

• No unified contract between stages — connecting a subscribable source to a pushable sink, a pullable buffer to a polling proxy, or a sync stream to an async consumer shouldn't require hand-written adapters every time.
• Type erosion across boundaries — as data passes through transformation, filtering, and async stages, input/output types are lost. Mismatches surface at runtime, not at compile time.
• Sync/async impedance mismatch — mixing synchronous reactive streams with async operations typically forces a full migration to one model or manual bridging at every boundary.
• Flow control as an afterthought — gating, routing, rate limiting, and fallback behavior are scattered across conditional flags and ad-hoc timers rather than composed from reusable primitives.
• Fragmented lifecycle management — subscriptions, timers, and intermediate nodes are cleaned up inconsistently. A uniform destroy() contract makes resource ownership explicit and leaks detectable.

Transferum provides composable, type-safe building blocks with a uniform capability system. You declare what each stage does (push, pull, transform, filter, poll) — the library handles how data moves between stages, including sync/async bridging, flow control, and resource cleanup.

Key Benefits

| Benefit | How | |-------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | Type-safe pipelines | Each transfer and operator carries its input/output types. Builders enforce type compatibility at compile time — a mismatch is a compile error, not a runtime crash. | | Uniform capability model | Every transfer declares its capabilities via flags (isPushable, isSubscribable, isGate, …). linkTransfers automatically selects the correct wiring strategy — no manual glue code. | | Sync + async in one system | Sync and async transfers coexist. linkTransfers prefers sync when possible and falls back to async strategies when needed. No separate "async world." | | Composable architecture | Transfers link into chains, bridges connect chains with gate control, builders assemble chains fluently, operators transform data — all orthogonal and reusable. | | Explicit lifecycle | Every resource (transfer, bridge, subscription, ticker) supports destroy(). Builders track owned resources and clean them up in one call. No leaked timers or subscriptions. | | Reactive by default, pull when needed | Most transfers are subscribable (push-based reactivity). Polling transfers add pull-based data acquisition on the same foundation. Use the right model per stage without switching libraries. | | Local, fail-safe error handling | Errors are local to each transfer — one stage's failure doesn't kill the pipeline. With onError — suppressed, stream continues. Without — visible (exception/rejection), and polling stops (no zombie tickers). Per-stage granularity (onAcceptError/onEmitError, onDestroyError). Typed ErrorHandler<TSource> passes the transfer instance. No silent swallowing. | | Undefined suppression | undefined never propagates through the chain of transfers — it means "no data", not "empty value." Use null as an explicit empty marker when needed. This eliminates an entire class of null-check bugs in downstream consumers. |

Use Cases

Real-time UI updates from API polling

import {
  OutputPipelineBuilder, createAsyncPollingSourceTransfer, createConvertTransfer, createMapOperator, createPushStoredChannelTransfer,
} from 'transferum';

// Poll an API every 5 seconds, transform the response, update subscribers
const polling = createAsyncPollingSourceTransfer<ServerState>({
  fetcher: async (): ServerState => await fetchApi('/api/state'),
  interval: 5000,
  activated: true,
});

const pipeline = OutputPipelineBuilder
  .start(polling)
  .to(createConvertTransfer<ServerState, ViewModel>({
    operator: createMapOperator((state) => toViewModel(state)),
  }))
  .finish(createPushStoredChannelTransfer<ViewModel>());

pipeline.subscribe((vm) => renderUI(vm));

Debounced user input with async validation

import {
  AsyncInputPipelineBuilder, createPushStoredChannelTransfer, createDebounceTransfer, createAsyncConditionTransfer,
  createAsyncConvertTransfer, createAsyncSinkTransfer, createAsyncMapOperator,
} from 'transferum';

// Debounce input → validate → transform → send to async sink
const input = createPushStoredChannelTransfer<string>();

const pipeline = AsyncInputPipelineBuilder
  .start(input)
  .to(createDebounceTransfer<string>({ delay: 300 }))
  .to(createAsyncConditionTransfer<string>({ shouldAccept: async (s) => s.length > 0 }))
  .to(createAsyncConvertTransfer<string, ValidationResult>({ operator: createAsyncMapOperator(async (s) => await validate(s)) }))
  .finish(createAsyncSinkTransfer<ValidationResult>({ callback: async (result) => await saveResult(result) }));

input.push('[email protected]'); // debounced → validated → saved

Merging multiple data sources into a single view

import { createAsyncPollingSourceTransfer, createPushStoredChannelTransfer, createMergeTransfer } from 'transferum';

// Merge multiple sensor streams into one (all sources must have the same type)
const tempSensor = createAsyncPollingSourceTransfer<SensorData>({
  fetcher: () => Promise.resolve({ sensor: 'temperature', value: 25 }),
  interval: 1000,
  activated: true,
});
const humiditySensor = createAsyncPollingSourceTransfer<SensorData>({
  fetcher: () => Promise.resolve({ sensor: 'humidity', value: 55 }),
  interval: 1000,
  activated: true,
});

const merge = createMergeTransfer<SensorData>({ sources: [tempSensor, humiditySensor] });
merge.subscribe((data) => updateDashboard(data)); // receives data from both sensors

Conditional routing with bridges

import { createBridgeSelector, createPassBridge } from 'transferum';

// Route data to different processing pipelines based on a selector
const fastBridge = createPassBridge({ source, target: fastPipeline, activated: false });
const slowBridge = createPassBridge({ source, target: slowPipeline, activated: false });

const router = createBridgeSelector({
  bridges: { fast: fastBridge, slow: slowBridge },
  initialKey: 'fast',
  activated: true,
});

// Switch route at runtime
router.select('slow');

Idle fallback polling

import { createAsyncIdlePollingTransfer } from 'transferum';

// When sensor push API stops interacting, fall back to polling for fresh data
const channel = createAsyncIdlePollingTransfer<SensorState>({
  fetcher: async () => await fetchSensorState(),
  timeout: 10000,   // 10 s of inactivity → start polling
  interval: 2000,   // poll every 2 s
  activated: true,
  onError: (e, currentChannel) => {
    console.warn('Cannot get sensor state');
    showAlert(currentChannel);
  },
});

channel.subscribe((item) => appendToChart(item));

// Sensor pushes items (e.g. by websocket) → real-time. User goes idle → automatic polling kicks in.
channel.push({ temperature: 25 });

Async data pipeline with storage

import {
  AsyncDuplexPipelineBuilder, createPushStoredChannelTransfer, createAsyncConvertTransfer, createAsyncMapOperator,
} from 'transferum';

// Push data → async transform → notify subscribers
const source = createPushStoredChannelTransfer<RawData>();

const pipeline = AsyncDuplexPipelineBuilder
  .start(source)
  .to(createAsyncConvertTransfer<RawData, ProcessedData>({
    operator: createAsyncMapOperator(async (raw) => await process(raw)),
  }))
  .finish(createPushStoredChannelTransfer<ProcessedData>());

pipeline.subscribe((data) => console.log('processed data', data));

source.push(rawData); // → async transformation → processed data

Broadcast to multiple consumers

import { createPushStoredChannelTransfer, createSplitTransfer, createSinkTransfer, createWriteTransfer, linkTransfers } from 'transferum';

// One source → multiple independent consumers
const source = createPushStoredChannelTransfer<Telemetry>();

const split = createSplitTransfer<Telemetry>({
  targets: [
    createSinkTransfer({ callback: (t) => logTelemetry(t) }),
    createSinkTransfer({ callback: (t) => updateChart(t) }),
    createWriteTransfer({ flow: telemetryStorage }),
  ],
});

linkTransfers(source, split);
source.push(telemetry); // → logged, charted, and stored simultaneously

Domain-Specific Applications

Transferum is designed for building complex, predictable data processing systems across various domains:

Game Development

Input Processing Pipeline

Collect events from keyboard → filter (e.g., DebounceTransfer to prevent spam) → transform into game commands → route to appropriate systems.

import {
  createDebounceTransfer, createConvertTransfer, createMapOperator,
  createBridgeSelector, createPassBridge,
} from 'transferum';

// Raw input source (for keystrokes)
// Delays handling by 16ms (~1 frame at 60 FPS) to debounce rapid inputs
const rawInputSource = createDebounceTransfer<KeyboardEvent>({ delay: 16 });

// Converter: transforms raw KeyboardEvent into a simple string action
const inputConverter = createConvertTransfer<KeyboardEvent, string>({
  operator: createMapOperator((event) => event.code), // Extracts the key code
});

// Target subsystems (consumers that process the final commands)
const carSystem = { push: (cmd: string) => console.log(`🚗 Car executing: ${cmd}`) };
const planeSystem = { push: (cmd: string) => console.log(`✈️ Plane executing: ${cmd}`) };
const menuSystem = { push: (cmd: string) => console.log(`📋 Menu processing: ${cmd}`) };

// Router: manages which input context/subsystem is currently active
const gameplayRouter = createBridgeSelector({
  bridges: {
    driving: createPassBridge({ source: inputConverter, target: carSystem }),
    flying: createPassBridge({ source: inputConverter, target: planeSystem }),
    ui: createPassBridge({ source: inputConverter, target: menuSystem }),
  },
  initialKey: 'driving', // The player starts inside a car by default
  activated: true,
});

// Pipe data from the raw source to the converter (proper subscription without loops)
rawInputSource.subscribe((event) => inputConverter.push(event));

// Gameplay Simulation:

async function runSimulation() {
  // 1. Player presses 'KeyW' while driving the car
  rawInputSource.push(new KeyboardEvent('keydown', { code: 'KeyW' }));
  // Expected Output after debounce: 🚗 Car executing: KeyW

  // Wait for debounce timer (16ms) to fire and deliver data to carSystem
  await sleep(20);

  // 2. Player boards a plane — switch the control context
  gameplayRouter.select('flying');

  // 3. Player presses the exact same 'KeyW' key
  rawInputSource.push(new KeyboardEvent('keydown', { code: 'KeyW' }));
  // Expected Output after debounce: ✈️ Plane executing: KeyW

  // Wait for the second debounce timer to fire
  await sleep(20);
}

runSimulation();

Particle & Sound Effects

Use BridgeMultiSelector to activate multiple effects simultaneously on events (explosion, hit).

import { createBridgeMultiSelector, createPassBridge } from 'transferum';

const effects = createBridgeMultiSelector({
  bridges: {
    explosion: createPassBridge({ source: trigger, target: particleSystem, activated: false }),
    sound: createPassBridge({ source: trigger, target: audioSystem, activated: false }),
    shake: createPassBridge({ source: trigger, target: cameraShake, activated: false }),
  },
  initialKeys: [],
  activated: true,
  owned: true,
});

// On explosion event
effects.check('explosion');
effects.check('sound');
effects.check('shake');

IoT & Automation

Sensor Data Aggregation

Read data from multiple sensors (temperature, humidity, motion) via AsyncPollingSourceTransfer → filter (ConditionTransfer) → aggregate → send to cloud or local storage.

import { OutputPipelineBuilder, createAsyncPollingSourceTransfer, createMergeTransfer, createConditionTransfer, createAsyncWriteTransfer } from 'transferum';

const sensor1 = createAsyncPollingSourceTransfer<SensorData>({
  fetcher: () => Promise.resolve({ temperature: 25, humidity: 50 }),
  interval: 50,
  activated: true,
});

const sensor2 = createAsyncPollingSourceTransfer<SensorData>({
  fetcher: () => Promise.resolve({ temperature: 26, humidity: 55 }),
  interval: 50,
  activated: true,
});

const aggregator = createMergeTransfer<SensorData>({
  sources: [sensor1, sensor2],
});

const pipeline = OutputPipelineBuilder
  .start(aggregator)
  .to(createConditionTransfer<SensorData>({ shouldAccept: (d) => d.temperature > 0 && d.humidity >= 0 }))
  .finish(createAsyncWriteTransfer<SensorData>({ flow: cloudStorage }));

Device Control

Process commands from users or external systems → route to specific actuators (BridgeSelector) → receive feedback.

import { createBridgeSelector, createPassBridge } from 'transferum';

const commandRouter = createBridgeSelector({
  bridges: {
    light: createPassBridge({ source: commandChannel, target: lightController, activated: false }),
    thermostat: createPassBridge({ source: commandChannel, target: thermostatController, activated: false }),
    lock: createPassBridge({ source: commandChannel, target: lockController, activated: false }),
  },
  initialKey: 'light',
  activated: true,
  owned: true,
});

commandRouter.select('thermostat'); // switch to thermostat control

Monitoring & Alerts

Poll device temperature via AsyncPollingSourceTransfer → filter by threshold (ConditionTransfer) → throttle alerts (ThrottleTransfer) → transform into alert (ConvertTransfer) → send notification.

import { OutputPipelineBuilder, createConditionTransfer, createThrottleTransfer, createConvertTransfer, createMapOperator, createAsyncPollingSourceTransfer } from 'transferum';

const TEMPERATURE_THRESHOLD = 95;

const tempMonitor = createAsyncPollingSourceTransfer<number>({
  fetcher: async () => await readTemperature(),
  interval: 1000,
  activated: true,
});

const alertPipeline = OutputPipelineBuilder
  .start(tempMonitor)
  .to(createConditionTransfer({ shouldAccept: (temp) => temp > TEMPERATURE_THRESHOLD }))
  .to(createThrottleTransfer({ interval: 5000 }))
  .finish(createConvertTransfer({ operator: createMapOperator((temp): Alert => ({ type: 'HIGH_TEMP', value: temp })) }));

alertPipeline.subscribe((alert) => sendNotification(alert));

UI/UX Applications

Reactive Forms

Process user input in form fields → DebounceTransfer for autosave or live search → validate (ConditionTransfer) → async transform (MapOperator) → store results.

import {
  AsyncDuplexPipelineBuilder, createDebounceTransfer, createAsyncConditionTransfer,
  createAsyncConvertTransfer, createPushStoredChannelTransfer, createAsyncMapOperator,
} from 'transferum';

const searchInput = createDebounceTransfer<string>({ delay: 300 });

const pipeline = AsyncDuplexPipelineBuilder
  .start(searchInput)
  .to(createAsyncConditionTransfer<string>({ shouldAccept: async (s) => s.length >= 3 }))
  .to(createAsyncConvertTransfer<string, SearchResult[]>({
    operator: createAsyncMapOperator(async (query) => await searchAPI(query)),
  }))
  .finish(createPushStoredChannelTransfer<SearchResult[]>());

pipeline.subscribe((results) => renderSuggestions(results));
searchInput.push('user query');

Monitoring & Logging Systems

Metrics Collection

Collect metrics from various sources (server logs, client events) → filter → transform → send to multiple monitoring systems (BridgeMultiSelector for Prometheus, ELK, Sentry simultaneously).

import { createPushStoredChannelTransfer, createBridgeMultiSelector, createPassBridge } from 'transferum';

const metricsChannel = createPushStoredChannelTransfer<Metric>();

const destinations = createBridgeMultiSelector({
  bridges: {
    prometheus: createPassBridge({ source: metricsChannel, target: prometheusWriter, activated: true }),
    elk: createPassBridge({ source: metricsChannel, target: elkWriter, activated: true }),
    sentry: createPassBridge({ source: metricsChannel, target: sentryWriter, activated: false }),
  },
  initialKeys: ['prometheus', 'elk'],
  activated: true,
  owned: true,
});

metricsChannel.push({ name: 'request_latency', value: 150 });

Financial Applications

Stock Market Data Processing

Receive streaming quotes → calculate indicators (MapOperator) → filter by conditions (ConditionTransfer) → transform into trading signals (ConvertTransfer) → execute trades.

import {
  DuplexPipelineBuilder, createPushChannelTransfer, createPushStoredChannelTransfer, createConvertTransfer,
  createConditionTransfer, createAsyncSinkTransfer, createMapOperator,
} from 'transferum';

const quoteStream = createPushChannelTransfer<Quote[]>();
const thresholdChannel = createPushStoredChannelTransfer<number>({ initialValue: 100 });

const indicatorPipeline = DuplexPipelineBuilder
  .start(quoteStream)
  .to(createConvertTransfer<Quote[], TechnicalIndicator>({
    operator: createMapOperator((quotes): TechnicalIndicator => ({
      value: quotes.reduce((sum, q) => sum + q.price, 0),
      symbols: quotes.map((q) => q.symbol),
      threshold: thresholdChannel.pull() ?? 0,
    })),
  }))
  .to(createConditionTransfer<TechnicalIndicator>({ shouldAccept: (ind) => ind.value > ind.threshold }))
  .to(createConvertTransfer<TechnicalIndicator, TradingSignal>({
    operator: createMapOperator((ind) => ({
      action: 'BUY',
      symbols: ind.symbols,
      targetPrice: ind.value,
    })),
  }))
  .finish(createAsyncSinkTransfer<TradingSignal>({ callback: async (signal) => await executeTrade(signal) }));

quoteStream.push([{ symbol: 'AAPL', price: 150, timestamp: Date.now() }]);
thresholdChannel.push(200);

Portfolio Management

Use BridgeSelector to switch between different strategies or data sources.

import { createBridgeSelector, createPassBridge } from 'transferum';

const strategyRouter = createBridgeSelector({
  bridges: {
    conservative: createPassBridge({ source: marketData, target: conservativeStrategy, activated: false }),
    aggressive: createPassBridge({ source: marketData, target: aggressiveStrategy, activated: false }),
    balanced: createPassBridge({ source: marketData, target: balancedStrategy, activated: true }),
  },
  initialKey: 'balanced',
  activated: true,
  owned: true,
});

// Switch strategy based on market conditions
if (marketVolatility > HIGH_THRESHOLD) {
  strategyRouter.select('conservative');
}

Comparison with Alternatives

Transferum exists in a rich ecosystem of reactive and stream-processing libraries. This section compares it with popular alternatives to help you make an informed choice.

Transferum vs RxJS

RxJS is the most widely adopted reactive programming library for JavaScript/TypeScript.

| Aspect | Transferum | RxJS | |----------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------| | Bundle size | ~15 KB minified | ~35 KB minified (full), <5 KB (selective imports) | | Dependencies | Zero | Zero (v7+) | | Learning curve | Moderate — explicit primitives | Steep — 100+ operators, complex concepts | | Type inference | Strong — tuple-based pipeline types | Strong — but complex generic chains | | Sync/Async unify | Built-in — linkTransfers handles both | Manual — from(), toPromise(), firstValueFrom() | | Pull-based | Native — Pullable, PollingProxy | Limited — mostly push-based | | Gate/Flow control | Built-in — GateTransfer, BridgeSelector | Manual — takeUntil(), switchMap(), subjects | | Resource cleanup | Explicit — destroy() on every transfer | Subscription-based — subscription.unsubscribe() | | Error handling | Local, non-fatal — onError per transfer, fail-safe polling, typed source | Stream-level — catchError(), retry(), errors terminate stream | | Undefined handling | Suppressed — undefined never propagates | Propagated — undefined is a valid value | | Operators | ~10 pure operators (stateless transforms only) | 100+ operators (creation, transformation, filtering, combination, utility) | | Flow control / Rate limiting | Built-in transfers — DebounceTransfer, ThrottleTransfer, BufferTransfer, GateTransfer (with explicit lifecycle) | Built-in operators — throttle(), buffer(), sample() (state lives in subscription) | | Operator-equivalent coverage | Many RxJS operators are transfers: debounceTime→DebounceTransfer, filter→ConditionTransfer, merge→MergeTransfer, share→SplitTransfer, takeUntil→GateTransfer, delay→DelayedPushChannelTransfer | All flow control is operator-based — no separate node lifecycle | | Testing | Simple — fake timers, direct method calls | Complex — TestScheduler, marble diagrams | | Community | Small — single maintainer | Large — Google, widespread adoption |

Key differences:

• Architecture: RxJS uses Observable + Operator + Subscription model. Transferum uses Transfer + Bridge + Builder with capability flags.
• Composability: RxJS operators are functions that transform observables. Transferum transfers are objects that can be linked via linkTransfers() automatically.
• Error handling: RxJS errors propagate through the stream and terminate it unless caught with catchError() / retry(). Transferum errors are local to each transfer — with onError, suppressed and the stream continues; without, the error is visible (exception/rejection) and polling stops (fail-safe). One stage's failure doesn't kill the pipeline. See Error Handling.
• Scheduling: RxJS has Scheduler abstraction (async, asyncSchedule, animationFrame). Transferum has Ticker (RAFTicker, IntervalTicker) for polling.

Code comparison — Conditional routing with runtime switching:

// RxJS
import { Subject, filter, mergeMap, from, catchError, EMPTY, Subscription } from 'rxjs';

const transports = {
  sentry: {
    level: 'ERROR',
    send: (l) => sentryAPI.send(l),
  },
  elk: {
    level: 'WARN',
    send: (l) => elkAPI.send(l),
  },
  prometheus: {
    level: 'INFO',
    send: (l) => prometheusAPI.send(l),
  },
};

// Subscription infrastructure
const source$ = new Subject<LogEntry>();
const subscriptions = new Map<string, Subscription>();

function activateRoute(key: string) {
  if (subscriptions.has(key)) return;

  const config = transports[key];
  if (!config) return; // Guard against unknown keys

  const sub = source$.pipe(
    // Filter logs by level from config
    filter((l) => l.level === config.level),

    // Wrap async send in Observable, swallow errors to keep source$ alive
    mergeMap((l) =>
      from(config.send(l)).pipe(
        catchError((e) => {
          console.error(`Error in ${key}:`, e);
          return EMPTY;
        })
      )
    )
  ).subscribe();

  subscriptions.set(key, sub);
}

function deactivateRoute(key: string) {
  subscriptions.get(key)?.unsubscribe();
  subscriptions.delete(key);
}

// Initial routes
activateRoute('sentry');
activateRoute('elk');
activateRoute('prometheus');

// Runtime: leave only Sentry and Prometheus
deactivateRoute('elk');

// Activate / deactivate single route
activateRoute('elk');
deactivateRoute('elk');

// Transferum
import {
  createBridgeMultiSelector, createPassBridge, createPushStoredChannelTransfer,
  createConditionTransfer, createAsyncSinkTransfer, createSplitTransfer, linkTransfers,
} from 'transferum';

const source = createPushStoredChannelTransfer<LogEntry>();

const sentryFilter = createConditionTransfer<LogEntry>({ shouldAccept: (l) => l.level === 'ERROR' });
const elkFilter = createConditionTransfer<LogEntry>({ shouldAccept: (l) => l.level === 'WARN' });
const prometheusFilter = createConditionTransfer<LogEntry>({ shouldAccept: (l) => l.level === 'INFO' });

linkTransfers(source, createSplitTransfer<LogEntry>({ targets: [sentryFilter, elkFilter, prometheusFilter] }));

const routes = createBridgeMultiSelector({
  bridges: {
    sentry: createPassBridge({
      source: sentryFilter,
      target: createAsyncSinkTransfer<LogEntry>({ callback: async (l) => sentryAPI.send(l), onError: (e) => console.error(e) }),
      activated: false,
    }),
    elk: createPassBridge({
      source: elkFilter,
      target: createAsyncSinkTransfer<LogEntry>({ callback: async (l) => elkAPI.send(l), onError: (e) => console.error(e) }),
      activated: false,
    }),
    prometheus: createPassBridge({
      source: prometheusFilter,
      target: createAsyncSinkTransfer<LogEntry>({ callback: async (l) => prometheusAPI.send(l), onError: (e) => console.error(e) }),
      activated: false,
    }),
  },
  initialKeys: ['sentry', 'elk', 'prometheus'],
  activated: true,
  owned: true,
});

// Runtime: leave only Sentry and Prometheus
routes.select(['sentry', 'prometheus']);

// Activate / deactivate single route
routes.check('elk');
routes.uncheck('elk');

RxJS handles routing via manual subscription management — a Map to track active subscriptions and explicit activateRoute/deactivateRoute functions. Transferum's BridgeMultiSelector is a first-class routing object: declarative bridge list, built-in subscription management (owned: true), and select() / check() / uncheck() to switch routes at runtime.

Code comparison — Debounced search with error handling and empty-result suppression:

The API call can fail — errors must be logged without killing the stream. Empty result sets must not reach the renderer.

// RxJS
import { fromEvent, from, EMPTY } from 'rxjs';
import { debounceTime, switchMap, filter, map, catchError } from 'rxjs/operators';

fromEvent(searchInput, 'input')
  .pipe(
    debounceTime(300),
    map((e: Event) => (e.target as HTMLInputElement).value),
    filter(query => query.length >= 3),
    switchMap(query =>
      from(searchAPI(query)).pipe(
        catchError(e => {
          console.error(e);
          return EMPTY;
        })
      )
    ),
    filter(results => results.length > 0)
  )
  .subscribe(results => render(results));

// Transferum
import {
  AsyncInputPipelineBuilder, createDebounceTransfer, createConditionTransfer,
  createAsyncConvertTransfer, createSinkTransfer, createAsyncMapOperator,
} from 'transferum';

const input = createDebounceTransfer<string>({ delay: 300 });

const pipeline = AsyncInputPipelineBuilder
  .start(input)
  .to(createConditionTransfer<string>({ shouldAccept: q => q.length >= 3 }))
  .to(createAsyncConvertTransfer<string, SearchResult[]>({
    operator: createAsyncMapOperator(async q => await searchAPI(q)),
    onError: (e) => console.error(e),
  }))
  .to(createConditionTransfer<SearchResult[]>({ shouldAccept: results => results.length > 0 }))
  .finish(createSinkTransfer<SearchResult[]>({
    callback: results => render(results),
  }), { owned: true });

input.push(query); // manually push, or integrate with DOM event

Transferum vs Most.js

Most.js is a lightweight, high-performance FRP library.

| Aspect | Transferum | Most.js | |-------------------|---------------------------------------------------|--------------------------| | Bundle size | ~15 KB | ~7 KB | | Status | Active (2026) | Maintenance mode (2020+) | | Async support | Built-in async transfers | Native async event loop | | Pull-based | Yes — Pullable, PollingProxy | No — push-only | | TypeScript | First-class, strict types | Community typings | | Operators | ~10 (pure transforms; flow control via transfers) | ~40 |

Most.js excels in raw performance for push-based streams but lacks Transferum's pull-based primitives and unified sync/async model.

Transferum vs Bacon.js / Kefir

Bacon.js and Kefir are Functional Reactive Programming (FRP) libraries with Property (stateful) and EventStream (stateless) abstractions.

| Aspect | Transferum | Bacon.js / Kefir | |--------------------|------------------------------------------------------|--------------------------------------| | State model | Explicit — ProxyReference<T> per transfer | Implicit — Property holds state | | Stream types | Capability flags (isSubscribable, isPullable) | Two types: EventStream, Property | | Error handling | Local, non-fatal — onError per transfer, fail-safe | Error events terminate stream | | Async | First-class async transfers | Via fromPromise() | | Bundle size | ~15 KB | ~12 KB (Bacon), ~8 KB (Kefir) | | Status | Active | Bacon: maintenance, Kefir: archived |

Transferum's capability flags provide more granularity than the two-type model, allowing fine-grained control over data flow mechanics.

Quick Comparison Table

| Feature | Transferum | RxJS | Most.js | Bacon.js | Kefir | |----------------------------|------------------------------|--------------------------------------|-------------------|-------------------|-------------------| | Bundle size (minified) | ~15 KB | ~35 KB | ~7 KB | ~12 KB | ~8 KB | | Dependencies | 0 | 0 | 0 | 0 | 0 | | Pull-based | ✓ | ✗ | ✗ | ✗ | ✗ | | Sync/Async unify | ✓ | Partial | Partial | Partial | Partial | | Built-in polling | ✓ (Ticker) | ✗ (manual) | ✗ | ✗ | ✗ | | Gate/Flow control | ✓ (GateTransfer, Bridge) | Manual | Manual | Manual | Manual | | Error handling | Local, non-fatal, fail-safe | Stream-level (catchError, retry) | Stream terminates | Stream terminates | Stream terminates | | Undefined suppression | ✓ | ✗ | ✗ | ✗ | ✗ | | Operators count | ~10 (pure transforms) | 100+ | ~40 | ~60 | ~50 | | Flow-control as nodes | ✓ (transfers with lifecycle) | ✗ (operators only) | ✗ | ✗ | ✗ | | TypeScript support | Excellent | Excellent | Good | Fair | Fair | | Community size | Small | Very large | Medium | Small | Small | | Maintenance status | Active | Active | Maintenance | Maintenance | Archived |

When to Choose Transferum

Transferum is ideal for:

1. TypeScript-first projects — Strict type inference, capability flags checked at compile time.
2. Mixed sync/async pipelines — Unified model without manual conversion.
3. Pull-based data acquisition — Polling APIs, sensors, or storage with PollingProxy.
4. Explicit flow control — Gates, bridges, and selectors for runtime routing.
5. Resource-conscious environments — Smaller bundle than RxJS, zero dependencies.
6. Resilient error handling — Local, non-fatal error handling: one stage's failure doesn't kill the pipeline, broken pollers stop (fail-safe), no silent swallowing. Per-stage granularity with typed ErrorHandler<TSource>.
7. Undefined-free data flow — undefined never propagates through the chain — it means "no data," not "empty value," eliminating null-check defects in consumers.
8. Game development / IoT — Frame-aligned tickers, idle polling, sensor aggregation.

Example fit: Real-time dashboard with API polling, debounced user input, conditional routing to multiple visualizations, and unified sync/async data flows.

When to Consider Alternatives

Consider RxJS if:

• You need operators not covered by Transferum's transfers: stream combinators (combineLatest, zip, withLatestFrom), windowing (bufferCount, windowTime), or retry policies (retryWhen). Many common operators (debounceTime, throttleTime, filter, merge, delay, takeUntil) have transfer equivalents — see the comparison table above.
• Your team already has RxJS expertise.
• You need advanced scheduling (test scheduler, virtual time).
• You need complex stream combination operators (e.g., combining 5+ distinct data sources with intricate join logic).
• Community support and long-term stability are top priorities.

Consider Most.js if:

• Raw performance is the top priority.
• You need a small, push-only stream library.
• You don't need pull-based or polling primitives.

Consider Bacon.js (or maintaining Kefir) only if:

• You are working on a legacy codebase already locked into these specific FRP abstractions.

Installation & Import

npm i transferum

All components are exported from a single entry point:

import {
  // Transfers
  PushChannelTransfer,
  DelayedPushChannelTransfer,
  DebounceTransfer,
  ThrottleTransfer,
  PushStoredChannelTransfer,
  BufferTransfer,
  ManualBufferTransfer,
  ManualFlowTransfer,
  GateTransfer,
  MergeTransfer,
  SplitTransfer,
  PollingSourceTransfer,
  PollingProxyTransfer,
  PollingFlowTransfer,
  IdlePollingTransfer,
  ChannelTransfer,
  StoredChannelTransfer,
  SinkTransfer,
  WriteTransfer,
  ReadTransfer,
  ConvertTransfer,
  ConditionTransfer,
  UniversalCompositeTransfer,

  // Async transfers
  AsyncSinkTransfer,
  AsyncWriteTransfer,
  AsyncReadTransfer,
  AsyncConvertTransfer,
  AsyncConditionTransfer,
  AsyncPollingSourceTransfer,
  AsyncPollingProxyTransfer,
  AsyncPollingFlowTransfer,
  AsyncIdlePollingTransfer,
  AsyncStoredChannelTransfer,

  // Operators
  TransparentOperator,
  MapOperator,
  FilterOperator,
  ReducerOperator,
  GuardOperator,
  PipelineOperator,

  // Async operators
  AsyncMapOperator,
  AsyncGuardOperator,
  AsyncPipelineOperator,

  // Storages
  LatestStorage,
  QueueStorage,
  StackStorage,

  // Tickers
  RAFTicker,
  IntervalTicker,
  TickerInterface,

  // Helpers
  Subscriber,
  SubscriptionManager,
  StateSubscriptionManager,
  ProxyReference,
  DisposableSubscriberAdapter,

  // Bridges
  PassBridge,
  TransformBridge,
  TransferBridge,
  BridgeAggregator,
  BridgeSelector,
  BridgeMultiSelector,
  AsyncTransformBridge,

  // Builders
  InputPipelineBuilder,
  OutputPipelineBuilder,
  DuplexPipelineBuilder,
  OperatorPipelineBuilder,

  // Async builders
  AsyncInputPipelineBuilder,
  AsyncOutputPipelineBuilder,
  AsyncDuplexPipelineBuilder,
  AsyncOperatorPipelineBuilder,

  // Factories
  createPushChannelTransfer,
  createDelayedPushChannelTransfer,
  createDebounceTransfer,
  createThrottleTransfer,
  createPushStoredChannelTransfer,
  // ... all create* functions (including createAsync*)

  // Utilities
  linkTransfers,
  handleError,
} from 'transferum';

Core Concepts

Capability Flags System

Each transfer implements CommunicationContractInterface — a set of boolean flags that determine available methods:

| Flag | Methods | Description | |-------------------|----------------------------------------------------------------------------------|--------------------------------------------------------------| | isInput | — | Can accept data from outside (acts as an input) | | isOutput | — | Can yield data (acts as an output) | | isDuplex | — | Both input and output simultaneously (isInput && isOutput) | | isPushable | push(data) | Data can be pushed into the transfer | | isPullable | pull() | Data can be read from the transfer | | isSubscribable | subscribe(handler) | The transfer can be subscribed to | | isTriggerable | trigger() | Has a manual emission trigger | | isGate | activate() / deactivate() / toggle() / active / onStateChange(handler) | Flow control (on/off) + state subscription | | isPollingSource | — | Has internal source polling | | isPollingProxy | setFetcher() / clearFetcher() | Polls the previous node in the chain |

Asynchronous flags:

| Flag | Methods | Description | |-----------------------|---------------------------------------------|-----------------------------------------------------| | isAsyncPushable | asyncPush(data) | Data can be pushed into the transfer asynchronously | | isAsyncPullable | asyncPull() | Data can be read from the transfer asynchronously | | isAsyncTriggerable | asyncTrigger() | Asynchronous manual emission trigger | | isAsyncPollingProxy | setAsyncFetcher() / clearAsyncFetcher() | Asynchronously polls the previous node in the chain |

isAsyncSubscribable and isAsyncPollingSource are not required — subscription remains synchronous in all async transfers, and isPollingSource is reused.

Flags are set as readonly properties in each transfer class. Flag checking occurs at runtime — calling a method not supported by the transfer throws an Error.

Sync priority over async: If a transfer supports both sync and async operations (e.g., UniversalCompositeTransfer with PushStoredChannelTransfer inside), linkTransfers prefers sync linking. Async strategies are applied only when sync is not applicable.

Transfers

A transfer is the fundamental building block of a pipeline. Each transfer:

1. Accepts data (push), yields data (pull), notifies subscribers (subscribe) — depending on its flags.
2. Manages internal state via ProxyReference<T>.
3. Supports a lifecycle: creation → operation → destroy().

Inheritance hierarchy:

BaseTransfer (abstract)
├── BaseStateTransfer<T> (abstract, adds ProxyReference<T>)
│   ├── PushChannelTransfer
│   ├── DelayedPushChannelTransfer
│   ├── DebounceTransfer
│   ├── ThrottleTransfer
│   ├── PushStoredChannelTransfer
│   ├── BufferTransfer
│   ├── ManualBufferTransfer
│   ├── ManualFlowTransfer
│   ├── GateTransfer
│   ├── MergeTransfer
│   ├── PollingSourceTransfer
│   ├── PollingProxyTransfer
│   ├── PollingFlowTransfer
│   ├── IdlePollingTransfer
│   ├── ChannelTransfer
│   ├── StoredChannelTransfer
│   ├── SinkTransfer
│   ├── ConvertTransfer
│   └── ConditionTransfer
├── SplitTransfer
├── WriteTransfer
└── ReadTransfer

UniversalCompositeTransfer (separate hierarchy, composition of input + output)

Linking Transfers

The linkTransfers(lhs, rhs) function connects an output transfer (LHS) to an input transfer (RHS), selecting a strategy based on flags:

| LHS | RHS | Strategy | |----------------------------------|----------------------------------|--------------------------------------------------------------------| | isSubscribable | isPushable | Reactive subscription: LHS notifies → RHS accepts | | isPullable | isPollingProxy | Active polling: RHS pulls data via setFetcher | | isSubscribable | isPollingProxy | Subscription + last-value buffering for the poller | | isSubscribable | isAsyncPushable | Subscription + asyncPush with .catch() (no ordering guarantee) | | isAsyncPullable | isAsyncPollingProxy | Active async polling: RHS pulls data via setAsyncFetcher | | isPullable | isAsyncPollingProxy | Sync-pull wrapped in an async fetcher | | isSubscribable | isAsyncPollingProxy | Subscription + buffer + async fetcher | | isAsyncPullable | isPollingProxy | Error — sync poller cannot await | | isPullable / isAsyncPullable | isPushable / isAsyncPushable | Error — a Bridge or Triggerable adapter is required | | other | other | Error — unsupported combination |

Sync priority: If both transfers support sync linking, it is used. Async strategies are applied only when sync is not applicable.

Rejection handling for subscribable → asyncPushable: .catch() is always called. If options.onError is provided — it is invoked with (error, target) via handleError() and the rejection is suppressed. Without onError — handleError() rethrows, resulting in an unhandled promise rejection (the source's subscription remains active). This is consistent with per-transfer error handling: without onError, errors are visible, not silently swallowed.

Ordering for subscribable → asyncPushable: No ordering guarantee — fast sync notifications from LHS can overtake pending asyncPush calls. A serializer is a separate task.

Returns SubscriberInterface for breaking the link.

import { createPushChannelTransfer, createSinkTransfer, linkTransfers } from 'transferum';

const source = createPushChannelTransfer<number>();
const target = createSinkTransfer<number>({ callback: (n) => console.log(n) });

const link = linkTransfers(source, target);
source.push(42); // → callback called

link.unsubscribe(); // break the link

For async links (subscribable → asyncPushable), you can pass options.onError to intercept rejections:

import { linkTransfers } from 'transferum';

const link = linkTransfers(source, asyncTarget, { onError: (e) => console.error(e) });

Undefined Behavior in Data Flows

undefined is never propagated through a transfer chain. If a value is undefined, subscribers are not notified — the event is fully suppressed at the SubscriptionManager.sendState() level.

This behavior is intentional: undefined means "no data" in the library, not "empty value." If you need to propagate an empty value, use null:

import { createPushStoredChannelTransfer } from 'transferum';

const channel = createPushStoredChannelTransfer<string | null>({ initialValue: null });

channel.subscribe((data) => console.log(data));

channel.push('hello');   // → "hello"
channel.push(null);      // → null (null MUST reach the subscriber)
channel.push(undefined); // → nothing happens (subscribers NOT notified)

If the data type allows null (e.g., string | null), use it as an explicit empty marker. For strings, this could be '' (empty string), for numbers — 0, for objects — {} or null.

Error Handling

Transferum uses a uniform error handling model across all transfers. Every transfer that can encounter a runtime error (fetcher failure, callback throw, flow read/write error, predicate throw) accepts an optional onError handler in its config.

handleError() and ErrorHandler<TSource>

type ErrorHandler<TSource> = (e: Error, source: TSource) => void;

function handleError<TSource>(error: unknown, source: TSource, onError?: ErrorHandler<TSource>): void;

• Non-Error values are converted to Error (via String(error)).
• If onError is provided — invokes it with (error, source) and suppresses the exception.
• If onError is not provided — rethrows the exception.
• If onError is provided but itself throws — that exception is rethrown (the handler is considered broken).

The source parameter is the transfer instance where the error occurred, allowing handlers to distinguish sources and access transfer state.

Three scenarios

| Scenario | onError provided | onError behavior | Result | |---|---|---|---| | Handler suppresses | ✓ | Returns normally | Exception suppressed, operation continues | | No handler | ✗ | — | Exception rethrown | | Handler throws | ✓ | Throws | Handler's exception rethrown |

Per-transfer error handlers

Most transfers use a single onError handler. Exceptions:

• ConditionTransfer / AsyncConditionTransfer — separate onAcceptError (for shouldAccept failures) and onEmitError (for shouldEmit failures), since the two stages are independent.
• ChannelTransfer / StoredChannelTransfer / AsyncStoredChannelTransfer — onError covers emit() failures; onDestroyError covers destroy() failures. setup() errors are always rethrown — a failed setup means the transfer is unusable, and suppressing would create a zombie object.

Polling transfers — error behavior

When a fetcher or flow read fails inside trigger() / asyncTrigger(), the error is passed to handleError(). The behavior differs depending on whether the error was suppressed:

Sync polling (PollingSourceTransfer, PollingProxyTransfer, PollingFlowTransfer, IdlePollingTransfer):

| Method | onError suppresses | onError absent or throws | |-------------|---------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------| | trigger() | Error suppressed, polling continues | Exception rethrown, ticker stops — polling ceases. Results in an uncaught exception (the ticker calls trigger() synchronously). | | pull() | Error suppressed, returns undefined | Exception rethrown to caller. Ticker is not affected. |

Async polling (AsyncPollingSourceTransfer, AsyncPollingProxyTransfer, AsyncPollingFlowTransfer, AsyncIdlePollingTransfer):

| Method | onError suppresses | onError absent or throws | |------------------|---------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------| | asyncTrigger() | Error suppressed, polling continues | Rejection rethrown, ticker stops — polling ceases. Results in an unhandled promise rejection (the ticker calls asyncTrigger() fire-and-forget). | | asyncPull() | Error suppressed, returns undefined | Rejection rethrown to caller. Ticker is not affected. |

Why the difference? trigger() / asyncTrigger() are called by the ticker (fire-and-forget), so an unhandled error surfaces as an uncaught exception (sync) or unhandled rejection (async). pull() / asyncPull() are called directly by the user, so the error propagates to the caller's try/catch or await expression.

Recommendation: always provide onError for polling transfers in production to prevent uncaught exceptions and unhandled rejections from stopping your application.

linkTransfers — async-push rejection

When linking a Subscribable source to an AsyncPushable target, asyncPush() rejections are caught and passed to handleError(). If options.onError is provided — it is invoked with (error, target) and the rejection is suppressed. Without onError — handleError() rethrows, resulting in an unhandled promise rejection. The source's subscription remains active in both cases — the error does not disrupt the reactive stream. See Linking Transfers.

Transfers

Transfer Comparison Table

| Transfer | Push | Pull | Sub | Trig | Gate | Poll | In | Out | Purpose | |----------------------------|:----:|:----:|:---:|:----:|:----:|:-------:|:--:|:---:|-----------------------------------------------------| | PushChannelTransfer | ✓ | — | ✓ | — | — | — | ✓ | ✓ | Reactive channel, data not retained | | DelayedPushChannelTransfer | ✓ | — | ✓ | — | — | — | ✓ | ✓ | Channel with delayed emission to subscribers | | DebounceTransfer | ✓ | — | ✓ | — | — | — | ✓ | ✓ | Channel with debounce emission (last after pause) | | ThrottleTransfer | ✓ | — | ✓ | — | — | — | ✓ | ✓ | Channel with throttle emission (leading + trailing) | | PushStoredChannelTransfer | ✓ | ✓ | ✓ | ✓ | — | — | ✓ | ✓ | Channel with last-value caching | | BufferTransfer | ✓ | ✓ | — | — | — | — | ✓ | ✓ | Passive buffer (push/pull without notifications) | | ManualBufferTransfer | ✓ | ✓ | — | ✓ | — | — | ✓ | ✓ | Buffer with read only after trigger() | | ManualFlowTransfer | ✓ | — | ✓ | ✓ | — | — | ✓ | ✓ | Emission to subscribers only on trigger() | | GateTransfer | ✓ | — | ✓ | — | ✓ | — | ✓ | ✓ | Flow blocking by state (active) | | MergeTransfer | — | — | ✓ | — | — | — | — | ✓ | Merge multiple sources | | SplitTransfer | ✓ | — | — | — | — | — | ✓ | — | Broadcast to multiple targets (broadcast) | | PollingSourceTransfer | — | ✓ | ✓ | ✓ | ✓ | Src | — | ✓ | Poll external fetcher on a timer | | PollingProxyTransfer | — | ✓ | ✓ | ✓ | ✓ | Src+Prx | ✓ | ✓ | Poll previous node on a timer | | PollingFlowTransfer | — | ✓ | ✓ | ✓ | ✓ | Src | — | ✓ | Poll from OutputFlowInterface (Storage) | | IdlePollingTransfer | ✓ | ✓ | ✓ | ✓ | ✓ | Src | ✓ | ✓ | Fallback polling on idle incoming data | | ChannelTransfer | — | — | ✓ | — | — | — | — | ✓ | External source via setup/destroy | | StoredChannelTransfer | — | ✓ | ✓ | ✓ | — | — | — | ✓ | Channel with storage + external source | | SinkTransfer | ✓ | — | — | — | — | — | ✓ | — | Terminal sink (callback) | | WriteTransfer | ✓ | — | — | — | — | — | ✓ | — | Write to InputFlowInterface (Storage) | | ReadTransfer | — | ✓ | — | — | — | — | — | ✓ | Read from OutputFlowInterface (Storage) | | ConvertTransfer | ✓ | — | ✓ | — | — | — | ✓ | ✓ | Transform via Operator | | ConditionTransfer | ✓ | — | ✓ | — | — | — | ✓ | ✓ | Conditional filtering (shouldAccept/shouldEmit) |

Legend: Push = isPushable, Pull = isPullable, Sub = isSubscribable, Trig = isTriggerable, Gate = isGate, Poll = polling (Src = isPollingSource, Prx = isPollingProxy), In = isInput, Out = isOutput.

UniversalCompositeTransfer is not included in the table — its flags are determined dynamically from the provided input and output transfers.

Async Transfer Comparison Table

| Transfer | aPush | aPull | aTrig | Sub | Gate | Poll | In | Out | Purpose | |----------------------------|:-----:|:-----:|:--