transferum
    Preparing search index...

    transferum

    Transferum

    npm jsr npm Coverage Status Build and test Minified Size License: MIT API Docs

    Transferum Logo

    A language for describing interactions between components.

    Type-safe primitives — transfers (nodes), bridges (edges), builders (chain constructors), and operators (data transformers) — compose into interaction graphs with compile-time-checked contracts.

    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.

    Not a stream library — a data transfer graph library. Unlike RxJS, where everything is an Observable and composition happens inside a single object, Transferum treats transfers as nodes and bridges as edges in a data transfer graph. linkTransfers connects nodes by inspecting their capability contracts — not their class names. Builders assemble subgraphs fluently. Operators are local graph transformations. This makes Transferum architecturally closer to dataflow systems and component graphs than to classical reactive stream libraries.

    Transfer ──── Bridge ──── Transfer
        \                        │
         \                       │
          ─────── Bridge ──── Transfer

    This is a graph.

    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. Flags also define the transfer's TypeScript interface at compile time — methods like push() or subscribe() are type-guaranteed, not runtime-guessed (see Capability Flags System).
    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 has an explicit cleanup method — destroy() for transfers, bridges, and subscription managers; stop() for tickers; unsubscribe() for subscriptions. 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.
    Built-in backpressure Four async transfers (AsyncSinkTransfer, AsyncWriteTransfer, AsyncConvertTransfer, AsyncConditionTransfer) support optional maxConcurrency, bufferSize, and onBufferOverflow — limiting parallel async operations, queuing excess data, and handling overflow gracefully. Defaults are backward-compatible (unlimited). See Backpressure.
    Ordered async execution AsyncSinkTransfer and AsyncWriteTransfer support optional ordered: true — callback/write invocations are executed sequentially in data-arrival order, regardless of their async duration. AsyncConvertTransfer and AsyncConditionTransfer automatically enforce ordered emission when maxConcurrency > 1 via an internal Sequence Guard (no config needed). See Backpressure.
    No god-objects, no utility sprawl BaseTransfer stays minimal — only capability declarations, no logic. Each transfer class models exactly one behavioral concept (buffering, gating, merging, polling...). No central object knows about every other component. Complexity is concentrated in the type layer; runtime code stays compact and readable.
    Pluggable linking LinkStrategyInterface lets you override how transfers are wired together — implement link() for custom logic (logging, validation, serialization, inter-process bridging) and inject it into CompositeTransferBuilder.start(transfer, { linkStrategy }). The default DefaultLinkStrategy delegates to linkTransfers() — zero-config for existing code, plug-and-play for custom needs. See Linking. 
    import {
      CompositeTransferBuilder, createAsyncPollingSourceTransfer, createConvertTransfer,
      createMapOperator, createPushStoredChannelTransfer,
    } from 'transferum';

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

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

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

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

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

    const pipeline = CompositeTransferBuilder
      .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)),
      }), { onLinkError: (e) => console.error(e) })
      .finish(createAsyncSinkTransfer<ValidationResult>({
        callback: async (result) => await saveResult(result),
      }), { onLinkError: (e) => console.error(e) });

    input.push('user@example.com'); // debounced → validated → saved

    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

    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,
      owned: false,
    });

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

    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 });

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

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

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

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

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

    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

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

    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,
      owned: 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');

    Sensor Data Aggregation

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

    import {
      CompositeTransferBuilder, 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 = CompositeTransferBuilder
      .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 {
      CompositeTransferBuilder, 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 = CompositeTransferBuilder
      .start(tempMonitor)
      .to(createConditionTransfer<number>({ shouldAccept: (temp) => temp > TEMPERATURE_THRESHOLD }))
      .to(createThrottleTransfer<number>({ interval: 5000 }))
      .finish(createConvertTransfer<number, Alert>({ operator: createMapOperator((temp): Alert => ({ type: 'HIGH_TEMP', value: temp })) }));

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

    Reactive Forms

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

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

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

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

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

    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: false }),
        elk: createPassBridge({ source: metricsChannel, target: elkWriter, activated: false }),
        sentry: createPassBridge({ source: metricsChannel, target: sentryWriter, activated: false }),
      },
      initialKeys: ['prometheus', 'elk'],
      activated: true,
      owned: true,
    });

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

    Stock Market Data Processing

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

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

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

    const indicatorPipeline = CompositeTransferBuilder
      .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),
      }), { onLinkError: (e) => console.error(e) });

    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');
    }

    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.

    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, DisplaceTransfer 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, switchMap→DisplaceTransfer 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.
    • Models, not abstractions: RxJS builds around a single abstraction (Observable) that serves as source, connection, and handler simultaneously. Transferum builds around distinct models — transfers are nodes, bridges are edges, operators are transforms. Each has a clear role; none tries to be all of them.
    • 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 {
      CompositeTransferBuilder, createPushChannelTransfer, createDebounceTransfer, createConditionTransfer,
      createDisplaceTransfer, createAsyncConvertTransfer, createAsyncMapOperator, createSinkTransfer,
    } from 'transferum';

    const input = createPushChannelTransfer<string>();

    const pipeline = CompositeTransferBuilder
      .start(input)
      .to(createDebounceTransfer<string>({ delay: 300 }))
      .to(createConditionTransfer<string>({ shouldAccept: q => q.length >= 3 }))
      .to(createDisplaceTransfer<string, SearchResult[]>({
        factory: () => createAsyncConvertTransfer<string, SearchResult[]>({
          operator: createAsyncMapOperator(async (query) => await searchAPI(query)),
          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

    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.

    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. Unlike Bacon.js/Kefir's runtime-only EventStream / Property distinction, Transferum's flags are compile-time type literals — TypeScript knows which methods each transfer exposes without runtime checks.

    Node Streams and WHATWG Streams (the web standard) both address data piping through typed stream objects.

    Aspect Transferum Node Streams / WHATWG Streams
    Type system Capabilities are composable — mix push + pull + gate + poll in one type 4 fixed roles (Readable / Writable / Duplex / Transform) — no composition of capabilities within a single stream
    Model Capabilities, not stream types Fixed stream types — a new role means a new class
    Push + Pull Both first-class — isPushable / isPullable as flags Readable = pull-oriented, Writable = push-oriented
    Gate / Routing Built-in — GateTransfer, BridgeSelector Manual — pipe chains, no routing primitive
    Error handling Local, non-fatal — per-transfer onError Stream-level — errors propagate and can destroy stream
    Lifecycle Explicit destroy() on every transfer destroy() / cancel() — inconsistent across impls
    TypeScript First-class — computed interfaces from flags Community typings (Node), limited generics (WHATWG)

    Node Streams and WHATWG Streams solve piping well, but they introduce a new stream type for each role (Readable, Writable, Duplex, Transform). Transferum introduces capabilities instead — a transfer declares what it can do via flags, and the type system computes the correct interface. Adding a new capability doesn't require a new stream class; it requires a new flag. This scales combinatorially better.

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

    Transferum is ideal for:

    1. TypeScript-first projects — Strict type inference. Capability flags are compile-time literals (isPushable: true, not boolean), so TypeScript knows which methods each transfer exposes — push(), subscribe(), pull() are type-guaranteed without casts or runtime checks.
    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.
    9. Infrastructure libraries — Where components interact in multiple ways (push, pull, polling, trigger, read, write) and the interaction contract matters more than the data format.
    10. Integration layers — Connecting heterogeneous components or services where a typed, capability-based contract is more valuable than a uniform stream interface.
    11. Complex pipelines with heterogeneous nodes — Where different stages have fundamentally different capabilities (some push-only, some pull-only, some gated, some polling) and these differences should be statically checked, not discovered at runtime.
    12. Embedded DSLs — Where the library serves as a language for describing an interaction graph with compile-time safety.

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

    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, switchMap) 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.

    Consider Node Streams / WHATWG Streams if:

    • You need the web standard or Node.js native streaming protocol for interop.
    • Your pipeline is purely I/O-oriented (file, network, compression) and doesn't need push+pull+gate+poll in one system.
    • You don't need compile-time capability guarantees.
    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,
      DisplaceTransfer,
      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,

      // Linking
      DefaultLinkStrategy,
      LinkStrategyInterface,

      // Builders
      InputPipelineBuilder,
      OutputPipelineBuilder,
      DuplexPipelineBuilder,
      OperatorPipelineBuilder,

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

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

      // Utilities
      linkTransfers,
      linkSubscribableToPushable,
      linkPullableToPollingProxy,
      linkSubscribableToPollingProxy,
      linkSubscribableToAsyncPushable,
      linkAsyncPullableToAsyncPollingProxy,
      linkPullableToAsyncPollingProxy,
      linkSubscribableToAsyncPollingProxy,
      handleError,

      // Guards
      isPushable,
      isPullable,
      isSubscribable,
      isPollingProxy,
      isTriggerable,
      isGate,
      isAsyncPushable,
      isAsyncPullable,
      isAsyncPollingProxy,
      isAsyncTriggerable,
    } from 'transferum';

    Transferum is built on a single idea:

    Behavior can be described as a composition of independent capabilities that simultaneously determine the type, the implementation, and the rules of interaction.

    Everything else — transfers, bridges, builders, operators — follows from this principle. They are its consequences, not the idea itself. The invariants below are the shape these consequences take in code.

    One idea, one direction. A single concept — capability — flows through every layer of the library:

    Capability flags
          ↓
    Type system (computed interfaces)
          ↓
    Transfer (declares capabilities, implements behavior)
          ↓
    Bridge (inspects capabilities, wires transfers)
          ↓
    Builder (assembles transfers, enforces type compatibility)

    Operators are not a separate layer — they are stateless transforms used inside transfers. Adding a new capability propagates automatically through all layers. Adding a new transfer class requires only declaring its flags — the type system, linkTransfers, and builders adapt without changes.

    The invariants:

    Invariant What it means Where enforced
    Transfers don't know their neighbors A transfer defines its own behavior (push, pull, subscribe, gate…) but never references, imports, or checks the class of another transfer. It doesn't know what's upstream or downstream — it only fulfills its own contract. BaseTransfer and all descendants — no cross-references between transfer classes
    Bridges don't know concrete implementations A bridge inspects capability flags, never class names. There is no instanceof chain, no RTTI, no class-name switching. Any transfer with the right capabilities is bridgeable — including ones that don't exist yet. linkTransfers, LinkStrategyInterface, PassBridge, BridgeSelector, all bridge classes
    Capabilities define the contract Flags are the single source of truth. They determine the TypeScript interface (compile-time), the available methods (runtime), the linking strategy (linkTransfers), and the builder type compatibility. One piece of metadata, many readers. interfaces.ts, types.ts, linkTransfers, LinkStrategyInterface, builders
    BaseTransfer is minimal The base class contains only capability declarations — no state, no logic. State lives in BaseStateTransfer<T> (one level down). This prevents the god-object pattern where a base accumulates knowledge of all descendants. BaseTransfer class hierarchy
    One class, one behavior Each transfer models exactly one behavioral concept. There are no combinatorial mega-classes (BufferedStoredGateTransfer). Complex behavior emerges from composition, not from inheritance depth. All transfer classes
    Operators are stateless transforms Operators transform data (apply(input) → output) and hold no pipeline state. They live inside transfers (ConvertTransfer, TransformBridge), not as standalone pipeline nodes. The transfer owns the behavioral contract; the operator owns the data transformation. OperatorInterface, AsyncOperatorInterface
    destroy() is universal Every resource has an explicit cleanup path — destroy() for transfers, bridges, and subscription managers; stop() for tickers; unsubscribe() for subscriptions. Builders track owned resources and clean them up in one destroy() call. No resource lacks a cleanup path. DisposableInterface, all transfers, all bridges, builders; TickerInterface.stop()
    undefined never propagates undefined means "no data," not "empty value." It is suppressed at SubscriptionManager.sendState() — subscribers are never notified with undefined. Use null for explicit empty markers. SubscriptionManager, all transfers

    Why these matter. A new transfer, bridge, or operator that respects these invariants integrates without touching existing code — the architecture stays coherent as it grows.

    Each transfer implements CommunicationContractInterface — a set of boolean flags. Each flag is both a runtime value and a compile-time guarantee: when a flag is true, the corresponding method is part of the transfer's TypeScript type — TypeScript knows it exists without any casts or runtime checks.

    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.

    How flags become compile-time guarantees: In interfaces.ts, each flag has a corresponding interface that narrows the flag to the literal type true (not boolean) and extends the matching method contract. When a transfer class implements these interfaces, TypeScript guarantees the presence of the methods at compile time:

    // Base contract — all flags are `boolean` (runtime-checkable)
    interface CommunicationContractInterface {
      readonly isPushable: boolean;
      readonly isPullable: boolean;
      readonly isSubscribable: boolean;
      // …
    }

    // isPushable: true  →  push(data: T) is guaranteed to exist
    interface PushableTransferInterface<T> extends PushableInterface<T>, BaseTransferInterface {
      readonly isInput: true;
      readonly isPushable: true;   // → TypeScript guarantees push(data) exists
    }

    // isPullable: true  →  pull(): T | undefined is guaranteed to exist
    interface PullableTransferInterface<T> extends PullableInterface<T>, BaseTransferInterface {
      readonly isOutput: true;
      readonly isPullable: true;   // → TypeScript guarantees pull() exists
    }

    // isSubscribable: true  →  subscribe(handler) is guaranteed to exist
    interface SubscribableTransferInterface<T> extends SubscribableInterface<T>, BaseTransferInterface {
      readonly isOutput: true;
      readonly isSubscribable: true;  // → TypeScript guarantees subscribe(handler) exists
    }

    This means: if a transfer's type includes isPushable: true, you can call push() on it — TypeScript will not error. If the flag is absent (or false), the method is not on the type, and calling it is a compile error, not a runtime surprise.

    This carries through to the type system in types.ts, where flags define entire transfer categories:

    • InputTransfer<T> — union of all interfaces with isInput: true (push, poll-proxy, gate, async-push, async-poll-proxy).
    • OutputTransfer<T> — union of all interfaces with isOutput: true (pull, subscribe, gate, async-pull).
    • DuplexTransfer<TIn, TOut> — intersection of InputTransfer<TIn> & OutputTransfer<TOut> with isDuplex: true.
    • Transfer<TIn, TOut, Features[]> — a computed type that resolves a list of feature flags into a concrete intersection of branded interfaces.
    • CompositeInputTransfer / CompositeOutputTransfer / CompositeDuplexTransfer — builder-produced types that expose only the capabilities of the underlying transfers, with optional triggerable and gate additions.

    Pipeline builders use these types to enforce capability compatibility at compile time — a mismatch (e.g., passing an output-only transfer where an input is required) is a type error.

    Why flags instead of class proliferation? Without capability flags, supporting every combination of push/pull/subscribe/read/write/gate would require a separate class for each combination (PushTransfer, PullTransfer, PushPullTransfer, PushSubTransfer, PushPullSubTransfer, …) — the class count grows combinatorially. Capability flags eliminate this explosion: one class declares its flags, and the type system computes the correct interface. This is the same principle behind Rust's Send/Sync/Copy traits and Go's small Reader/Writer/Closer interfaces — capabilities as composable metadata, not as inheritance branches.

    Beyond reactive: ECS and trait analogies. The capability model has structural parallels outside reactive programming. In Entity Component System (ECS), entities are bare containers, components are data tags, and systems operate on entities with specific components — capabilities resemble components, transfers resemble entities, bridges resemble systems. In Rust, traits (Send, Sync, Clone) constrain behavior without inheritance — Transferum's flags do the same: they describe what a transfer can do as composable metadata, not what it is in a class hierarchy.

    Single source of truth — not an SRP violation. Capability flags serve multiple consumers (type computation, interface narrowing, bridge strategy selection, operator dispatch). Yet they do not violate the Single Responsibility Principle: flags are metadata — they describe what a transfer can do and nothing else. The consumers that act on this metadata (type system, linkTransfers, builders) are separate subsystems. One piece of information, many readers — a single source of truth, not a single responsibility overload.

    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.

    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().

    One class, one behavior. Each transfer models exactly one behavioral concept — BufferTransfer buffers, GateTransfer gates, MergeTransfer merges, ConditionTransfer filters. There is no BufferedStoredGateTransfer or other combinatorial mega-class. Complex behavior emerges from composition (linking, bridges, builders), not from inheritance depth.

    Transfer ≠ State. The inheritance hierarchy is intentionally two-level: BaseTransfer declares capabilities only — no state, no logic. BaseStateTransfer<T> adds ProxyReference<T> for transfers that need to hold data. This separation keeps the base class minimal and avoids the god-object pattern where a single base accumulates state, logic, and knowledge of all descendants.

    Read/Write and Push/Pull are first-class distinctions, not implementation details. In RxJS, Observable conflates source, connection, and handler into one object. Transferum separates them: ReadTransfer and WriteTransfer embody a CQRS-like split where reading and writing are distinct contracts. Similarly, push and pull are fundamental behavioral characteristics — not interchangeable mechanics. Pull-based transfers get backpressure virtually for free; push-based transfers require explicit flow control. This distinction shapes the entire capability system.

    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
    │   └── DisplaceTransfer
    ├── SplitTransfer
    ├── WriteTransfer
    └── ReadTransfer

    UniversalCompositeTransfer (separate hierarchy, composition of input + output)

    The linkTransfers(lhs, rhs) function connects an output transfer (LHS) to an input transfer (RHS). It delegates to DefaultLinkStrategy.link(), 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.

    Contract-based linking, not class-based. linkTransfers never asks "what class is this?" — it asks "what capabilities does it have?" There is no instanceof chain, no RTTI, no class-name switching. The function inspects capability flags and dispatches to the matching strategy. This is protocol-oriented design: any transfer that declares the right capabilities is linkable, including new ones you create — linkTransfers does not need to change. Adding a new transfer class requires only declaring its flags; the linking machinery adapts automatically. This is the practical payoff of the capability system: the same flags that guarantee compile-time method existence also drive runtime wiring.

    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) });

    LinkStrategyInterface is a strategy for linking transfers:

    • link(lhs, rhs, options?) — directly connects an output transfer to an input transfer (same as linkTransfers())

    DefaultLinkStrategy is the standard implementation: link() inspects capability flags on both transfers and dispatches to the matching sync or async strategy. It is the zero-config default — existing code works unchanged.

    import { DefaultLinkStrategy } from 'transferum';

    const linkStrategy = new DefaultLinkStrategy();

    // Direct linking — equivalent to linkTransfers(source, target)
    const link = linkStrategy.link(source, target);

    For builder-based pipelines, pass a link strategy to CompositeTransferBuilder.start():

    import {
      CompositeTransferBuilder, DefaultLinkStrategy, createPushStoredChannelTransfer,
      createConditionTransfer, createSinkTransfer,
    } from 'transferum';

    const linkStrategy = new DefaultLinkStrategy();
    const pipeline = CompositeTransferBuilder
      .start(createPushStoredChannelTransfer<number>(), { linkStrategy })
      .to(createConditionTransfer<number>({ shouldAccept: x => x > 0 }))
      .finish(createSinkTransfer<number>({ callback: console.log }));

    pipeline.push(42); // → 42

    Implement LinkStrategyInterface to customize linking behavior — e.g., logging every link, validating combinations, serializing links across a network, or wrapping linkTransfers() with additional error handling. The builder calls linkStrategy.link() for every to() and finish(), so a single strategy instance controls the entire pipeline's wiring.

    import type { InputTransfer, LinkConfig, LinkStrategyInterface, OutputTransfer, SubscriberInterface } from 'transferum';
    import { linkTransfers } from 'transferum';

    class LoggingLinkStrategy implements LinkStrategyInterface {
      link<T, RTransfer extends InputTransfer<T>>(
        lhs: OutputTransfer<T>,
        rhs: RTransfer,
        options?: LinkConfig<RTransfer>,
      ): SubscriberInterface {
        console.log(`Linking ${lhs.constructor.name} → ${rhs.constructor.name}`);
        return linkTransfers(lhs, rhs, options);
      }
    }

    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 | undefined>({ 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.

    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.

    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.

    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

    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.

    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.

    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)
    DisplaceTransfer ✓ — ✓ — — — ✓ ✓ Switch-map: new inner per value, previous displaced

    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.

    Transfer aPush aPull aTrig Sub Gate Poll In Out BP Purpose
    AsyncSinkTransfer ✓ — — — — — ✓ — ✓ Async terminal sink (callback)
    AsyncWriteTransfer ✓ — — — — — ✓ — ✓ Async write to AsyncInputFlowInterface
    AsyncReadTransfer — ✓ — — — — — ✓ — Async read from AsyncOutputFlowInterface
    AsyncConvertTransfer ✓ — — ✓ — — ✓ ✓ ✓ Async transform via AsyncOperator
    AsyncConditionTransfer ✓ — — ✓ — — ✓ ✓ ✓ Async conditional filtering (async predicates)
    AsyncPollingSourceTransfer — ✓ ✓ ✓ ✓ Src — ✓ — Async poll external fetcher on a timer
    AsyncPollingProxyTransfer — ✓ ✓ ✓ ✓ Src+aPrx ✓ ✓ — Async poll previous node on a timer
    AsyncPollingFlowTransfer — ✓ ✓ ✓ ✓ Src — ✓ — Async poll from AsyncOutputFlowInterface
    AsyncIdlePollingTransfer ✓* ✓ ✓ ✓ ✓ Src ✓ ✓ — Fallback async polling on idle incoming data
    AsyncStoredChannelTransfer — ✓ ✓ ✓ — — — ✓ — Channel with storage + external source + async interface

    Legend: aPush = isAsyncPushable, aPull = isAsyncPullable, aTrig = isAsyncTriggerable, aPrx = isAsyncPollingProxy, BP = Backpressure (maxConcurrency / bufferSize / onBufferOverflow). ✓* — method is synchronous (push), but fetcher is asynchronous.

    Subscription in all async transfers remains synchronous — subscribe() notifies subscribers synchronously, even if data is obtained via asyncPush/asyncPull/asyncTrigger.

    Groups by purpose:

    Group Transfers Common characteristic
    Channels PushChannelTransfer, DelayedPushChannelTransfer, DebounceTransfer, ThrottleTransfer, PushStoredChannelTransfer Reactive data delivery to subscribers
    Buffers BufferTransfer, ManualBufferTransfer Synchronous data exchange without reactivity
    Controlled flows ManualFlowTransfer, GateTransfer Control of emission timing or condition
    Polling PollingSourceTransfer, PollingProxyTransfer, PollingFlowTransfer, IdlePollingTransfer Periodic source polling
    Adapters SinkTransfer, WriteTransfer, ReadTransfer Integration with external flows and storages
    Transformation ConvertTransfer, ConditionTransfer, DisplaceTransfer Data processing, filtering, and switch-mapping
    Aggregation MergeTransfer, SplitTransfer Merging and splitting flows
    External sources ChannelTransfer, StoredChannelTransfer Integration via setup/destroy callbacks
    Composition UniversalCompositeTransfer Combining input + output into a single interface
    Async adapters AsyncSinkTransfer, AsyncWriteTransfer, AsyncReadTransfer Async integration with external flows and storages
    Async transformation AsyncConvertTransfer, AsyncConditionTransfer Async data processing and filtering
    Async polling AsyncPollingSourceTransfer, AsyncPollingProxyTransfer, AsyncPollingFlowTransfer, AsyncIdlePollingTransfer Async periodic source polling
    Async external sources AsyncStoredChannelTransfer Integration via setup/destroy with async interface

    Reactive channel with automatic emission to subscribers on push(). Data is not retained after emission.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    import { createPushChannelTransfer } from 'transferum';

    const channel = createPushChannelTransfer<number>();

    channel.subscribe((data) => console.log(data));
    channel.push(42); // → 42
    // After push() state is cleared

    Reactive channel with delayed emission to subscribers on push(). Each push() schedules its own timer for delay ms, upon expiration of which data is sent to subscribers and state is cleared. Multiple push() calls create independent delayed notifications.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    import { createDelayedPushChannelTransfer } from 'transferum';

    const channel = createDelayedPushChannelTransfer<number>({ delay: 100 });

    channel.subscribe((data) => console.log(data));
    channel.push(42); // → 42 will be logged after 100 ms

    // Multiple pushes — independent timers
    channel.push(1);
    channel.push(2);
    // after 100 ms: → 1, → 2

    destroy() clears all pending timers — delayed notifications are canceled.

    Reactive channel with debounced emission to subscribers on push(). Each push() resets the previous timer; subscribers are notified only after delay ms of silence following the last push(). Only the last value is emitted.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    import { createDebounceTransfer } from 'transferum';

    const channel = createDebounceTransfer<number>({ delay: 200 });

    channel.subscribe((data) => console.log(data));
    channel.push(1); // resets timer
    channel.push(2); // resets timer
    // after 200 ms of silence → 2

    // Rapid push bursts — only the last value
    channel.push(10);
    channel.push(20);
    channel.push(30);
    // after 200 ms → 30

    destroy() cancels the pending timer — the delayed notification will not fire.

    Reactive channel with throttled emission to subscribers on push(). The first push() passes immediately (leading edge), subsequent ones within interval are ignored, but the last value is emitted after the interval ends (trailing edge). No value is lost.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    import { createThrottleTransfer } from 'transferum';

    const channel = createThrottleTransfer<number>({ interval: 100 });

    channel.subscribe((data) => console.log(data));
    channel.push(1); // → 1 (leading edge, immediately)
    channel.push(2); // saved as pending
    channel.push(3); // saved as pending (overwrote 2)
    // after 100 ms → 3 (trailing edge)

    // After the interval — new leading edge
    channel.push(4); // → 4 (immediately)

    destroy() cancels the pending timer and clears pending — the trailing notification will not fire.

    Reactive channel with last-value retention. The value is available for pull() after push().

    Capabilities: isInput, isOutput, isDuplex, isPushable, isPullable, isSubscribable, isTriggerable

    import { createPushStoredChannelTransfer } from 'transferum';

    const channel = createPushStoredChannelTransfer<number>({ initialValue: 0 });

    channel.subscribe((data) => console.log(data));
    channel.push(42); // → subscribers notified, value retained
    console.log(channel.pull()); // 42

    channel.trigger(); // re-emit current value to subscribers

    Passive buffer with push/pull mechanics (no notifications). pull() extracts the value with cleanup.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isPullable

    import { createBufferTransfer } from 'transferum';

    const buffer = createBufferTransfer<number>();

    buffer.push(42);
    console.log(buffer.pull()); // 42
    console.log(buffer.pull()); // undefined (buffer empty)

    Buffer with manual read control via trigger(). pull() returns data only after trigger().

    Capabilities: isInput, isOutput, isDuplex, isPushable, isPullable, isTriggerable

    import { createManualBufferTransfer } from 'transferum';

    const buffer = createManualBufferTransfer<number>();

    buffer.push(42);
    console.log(buffer.pull()); // undefined (trigger not called)

    buffer.trigger();
    console.log(buffer.pull()); // 42

    Reactive stream with manual emission control. push() writes the value, trigger() emits to subscribers.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable, isTriggerable

    import { createManualFlowTransfer } from 'transferum';

    const flow = createManualFlowTransfer<number>();

    flow.subscribe((data) => console.log(data));
    flow.push(42); // subscribers NOT notified
    flow.trigger(); // → 42

    Transfer with state management (gate). Passes data only when active === true.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable, isGate

    import { createGateTransfer } from 'transferum';

    const gate = createGateTransfer<number>({ activated: false });

    gate.subscribe((data) => console.log(data));
    gate.push(42); // ignored (gate closed)

    gate.activate();
    gate.push(100); // → 100

    gate.deactivate();
    gate.push(200); // ignored

    console.log(gate.toggle()); // true (open)

    Subscribing to state changes:

    import { createGateTransfer } from 'transferum';

    const gate = createGateTransfer<number>({ activated: false });

    gate.onStateChange((g) => {
      console.log(`Gate state changed: active=${g.active}`);
    });

    gate.activate();   // → "Gate state changed: active=true"
    gate.deactivate(); // → "Gate state changed: active=false"

    onStateChange() returns SubscriberInterface for unsubscription. The subscriber receives GateInterface (the transfer itself) in the callback.

    Aggregator of multiple sources into a single stream. Automatically subscribes to all sources.

    Capabilities: isOutput, isPushable, isSubscribable

    import { createPushStoredChannelTransfer, createMergeTransfer } from 'transferum';

    const source1 = createPushStoredChannelTransfer<number>();
    const source2 = createPushStoredChannelTransfer<number>();

    const merge = createMergeTransfer<number>({ sources: [source1, source2] });

    merge.subscribe((data) => console.log(data));
    source1.push(1); // → 1
    source2.push(2); // → 2

    Stream splitter to multiple targets (broadcast). push() sends data to all targets.

    Capabilities: isInput, isPushable

    import { createPushStoredChannelTransfer, createSplitTransfer } from 'transferum';

    const target1 = createPushStoredChannelTransfer<number>();
    const target2 = createPushStoredChannelTransfer<number>();

    const split = createSplitTransfer<number>({ targets: [target1, target2] });

    split.push(42); // sent to target1 and target2

    Output transfer with internal polling of a data source. A Ticker calls trigger() at a specified interval.

    Capabilities: isOutput, isPollingSource, isPullable, isSubscribable, isTriggerable, isGate

    import { createPollingSourceTransfer } from 'transferum';

    const polling = createPollingSourceTransfer<number>({
      fetcher: () => Date.now(),
      interval: 1000,
      activated: true,
    });

    polling.subscribe((data) => console.log(data));
    // Every second: current time

    polling.deactivate(); // stop polling

    Error handling: If fetcher() throws in trigger(), onError is called. With onError — suppressed, polling continues. Without onError (or if onError itself throws) — exception rethrown, ticker stops. If fetcher() throws in pull() — same logic, but the ticker is not affected (error propagates to caller).

    import { createPollingSourceTransfer } from 'transferum';

    const polling = createPollingSourceTransfer<number>({
      fetcher: () => { throw new Error('fail'); },
      interval: 1000,
      activated: true,
      onError: (e, source) => console.error(e, source.active),
    });

    All polling transfers (PollingSourceTransfer, PollingProxyTransfer, PollingFlowTransfer, IdlePollingTransfer) also support onStateChange() for subscribing to active state changes.

    Duplex transfer with polling that receives its fetcher from the previous node in the chain. The fetcher is set via setFetcher() (usually called by linkTransfers).

    Capabilities: isInput, isOutput, isDuplex, isPollingProxy, isPollingSource, isPullable, isSubscribable, isTriggerable, isGate

    import { createPollingProxyTransfer } from 'transferum';

    const poller = createPollingProxyTransfer<number>({
      interval: 1000,
      activated: false,
    });

    // fetcher is set via linkTransfers or manually:
    poller.setFetcher(() => someValue);

    poller.activate(); // start polling

    Error handling: Same as PollingSourceTransfer — onError suppresses fetcher errors in trigger() and pull(). Without onError (or if it throws) — trigger() rethrows and ticker stops; pull() rethrows to caller without affecting the ticker.

    Output transfer with polling from OutputFlowInterface (e.g., Storage).

    Capabilities: isOutput, isPollingSource, isPullable, isSubscribable, isTriggerable, isGate

    import { createLatestStorage, createPollingFlowTransfer } from 'transferum';

    const storage = createLatestStorage<number>(0);
    const polling = createPollingFlowTransfer<number>({
      flow: storage,
      interval: 1000,
      activated: true,
    });

    polling.subscribe((data) => console.log(data));
    storage.write(42); // after interval → 42

    Error handling: If flow.read() throws in trigger(), onError is called. With onError — suppressed, polling continues. Without onError (or if it throws) — trigger() rethrows, ticker stops. pull() rethrows to caller without affecting the ticker.

    Reactive channel with fallback polling on idle. If no data arrived via push() for longer than timeout ms, periodic polling of fetcher starts with interval ms. When new data arrives, polling stops and the idle timer resets.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isPullable, isSubscribable, isPollingSource, isTriggerable, isGate

    import { createIdlePollingTransfer } from 'transferum';

    const channel = createIdlePollingTransfer<number>({
      fetcher: () => fetchLatest(),
      timeout: 5000,   // 5 seconds without push → start polling
      interval: 1000,  // poll fetcher every second
      activated: true,
    });

    channel.subscribe((data) => console.log(data));
    channel.push(42);  // → subscribers notified, idle timer reset
    // after 5 seconds without push → polling fetcher every 1 second

    Error handling: If fetcher() throws during polling, onError is called. With onError — suppressed, polling continues. Without onError (or if it throws) — exception rethrown, polling stops. pull() rethrows to caller without affecting polling.

    Output channel with external management via setup/destroy callbacks. Used for integration with external event sources.

    Capabilities: isOutput, isSubscribable

    Error handling: setup() errors are always rethrown (no onSetupError) — a failed setup means the transfer is unusable. onError covers emit() failures. onDestroyError covers destroy() failures. Without the corresponding handler — rethrown.

    import { createChannelTransfer } from 'transferum';

    const channel = createChannelTransfer<number>({
      setup: (emit) => {
        const id = setInterval(() => emit(Date.now()), 1000);
      },
      destroy: () => {
        // resource cleanup
      },
      onError: (e) => console.error(e),
    });

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

    Channel with last-value retention and external management. The value is available for pull() and trigger().

    Capabilities: isOutput, isPullable, isTriggerable, isSubscribable

    Error handling: Same as ChannelTransfer — setup() errors always rethrown. onError covers emit() / trigger() failures. onDestroyError covers destroy() failures.

    import { createStoredChannelTransfer } from 'transferum';

    let emit: (data: number) => void;

    const channel = createStoredChannelTransfer<number>({
      setup: (e) => { emit = e; },
      destroy: () => {},
      initialValue: 0,
    });

    channel.subscribe((data) => console.log(data));
    emit(42); // → 42
    console.log(channel.pull()); // 42

    channel.trigger(); // re-emit

    Terminal destination — calls a callback on receiving data.

    Capabilities: isInput, isPushable

    import { createSinkTransfer } from 'transferum';

    const sink = createSinkTransfer<number>({
      callback: (data) => console.log('Received:', data),
    });

    sink.push(42); // → "Received: 42"

    Write adapter for an arbitrary InputFlowInterface (e.g., Storage).

    Capabilities: isInput, isPushable

    import { createLatestStorage, createWriteTransfer } from 'transferum';

    const storage = createLatestStorage<number>();
    const writer = createWriteTransfer<number>({ flow: storage });

    writer.push(42); // storage.write(42)

    Read adapter for an arbitrary OutputFlowInterface (e.g., Storage).

    Capabilities: isOutput, isPullable

    import { createLatestStorage, createReadTransfer } from 'transferum';

    const storage = createLatestStorage<number>();
    storage.write(42);
    const reader = createReadTransfer<number>({ flow: storage });

    console.log(reader.pull()); // 42

    Converter transfer: transforms input data via an Operator and sends the result to subscribers. If the operator returns undefined, subscribers are not notified.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    import { createConvertTransfer, createMapOperator } from 'transferum';

    const converter = createConvertTransfer<number, string>({
      operator: createMapOperator((n: number) => `val_${n}`),
    });

    converter.subscribe((data) => console.log(data));
    converter.push(42); // → "val_42"

    Transfer with conditional filtering on input (shouldAccept) and output (shouldEmit).

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    import { createConditionTransfer } from 'transferum';

    const condition = createConditionTransfer<number>({
      shouldAccept: (n) => n > 0,        // input filter
      shouldEmit: (n) => n < 100, // output filter
    });

    condition.subscribe((data) => console.log(data));
    condition.push(-5);  // rejected by shouldAccept
    condition.push(50);  // passed both filters → 50
    condition.push(150); // passed shouldAccept, rejected by shouldEmit

    Switch-map transfer: for each input value, creates a new inner async-pushable + subscribable transfer via a factory function, pushes the value into it via asyncPush(), and forwards the inner's emissions to outer subscribers. On each new push(), the previous inner subscription is unsubscribed and the previous inner transfer is destroyed — only the latest inner's emissions pass through.

    The outer push() is synchronous. The factory receives no arguments — it is purely declarative. DisplaceTransfer handles data delivery by calling inner.asyncPush(data) internally (fire-and-forget). The async work happens inside the inner transfer; results arrive via subscription callbacks.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable

    Configuration:

    • factory: () => Transfer<TInput, TOutput, [AsyncPushable, Subscribable]> — creates a new inner transfer per input value (no arguments; data is pushed via asyncPush)
    • onError?: ErrorHandler<DisplaceTransfer> — suppresses factory errors; without it, factory errors are rethrown
    • onDisplace?: (displaced: Transfer<TInput, TOutput, [AsyncPushable, Subscribable]>) => void — called with the previous inner transfer before it is unsubscribed and destroyed on displacement by a new push(). Use for custom cleanup that must happen before destruction (e.g., aborting an in-flight request, closing a WebSocket, cancelling a timer). Not called on destroy() — only on displacement by a new push(). Not called for the first push() (no previous inner exists).

    RxJS equivalent: switchMap

    import {
      createDisplaceTransfer, createAsyncConvertTransfer, createAsyncMapOperator,
    } from 'transferum';

    const displace = createDisplaceTransfer<string, SearchResult>({
      factory: () => createAsyncConvertTransfer<string, SearchResult>({
        operator: createAsyncMapOperator(async (query) => await searchApi(query)),
      }),
    });

    displace.subscribe((results) => render(results));

    displace.push('hello'); // creates inner, pushes 'hello' via asyncPush
    displace.push('world'); // displaces previous inner, creates new one

    onDisplace — custom cancellation before inner is destroyed:

    The onDisplace callback receives the previous inner transfer typed as the exact type returned by factory, so to abort an in-flight request you need a custom inner transfer that exposes an abort() method (a plain AsyncConvertTransfer doesn't have one).

    import { Transfer, AsyncPushable, Subscribable, BaseTransfer } from 'transferum';
    import { createDisplaceTransfer, createAsyncConvertTransfer, createAsyncMapOperator } from 'transferum';

    // Custom inner transfer: wraps AsyncConvertTransfer and exposes abort()
    class FetchTransfer extends BaseTransfer implements Transfer<string, SearchResult, [AsyncPushable, Subscribable]> {
      public readonly isInput = true;
      public readonly isOutput = true;
      public readonly isDuplex = true;

      public readonly isAsyncPushable = true;
      public readonly isSubscribable = true;

      private readonly _controller = new AbortController();
      private readonly _inner = createAsyncConvertTransfer<string, SearchResult>({
        operator: createAsyncMapOperator(async (query) => {
          return (await fetch(`/api/search?q=${query}`, { signal: this._controller.signal })).json();
        }),
      });

      abort() {
        this._controller.abort();
      }
      asyncPush(data: string) {
        return this._inner.asyncPush(data);
      }
      subscribe(handler: (data: SearchResult) => void) {
        return this._inner.subscribe(handler);
      }
      destroy() {
        this._inner.destroy();
      }
      // ...delegate remaining capability flags/methods
    }

    const displace = createDisplaceTransfer<string, SearchResult, FetchTransfer>({
      factory: () => new FetchTransfer(),
      onDisplace: (displaced) => {
        // displaced is typed as FetchTransfer — no cast needed.
        // Abort the in-flight fetch so it doesn't waste resources.
        displaced.abort();
      },
    });

    displace.push('hello'); // starts fetch
    displace.push('world'); // onDisplace aborts 'hello' fetch, then destroys the inner

    Use cases:

    • Search-as-you-type: debounce → displace(factory) → latest result wins
    • Per-value async operations (fetch, readFile) where only the latest result matters
    • Per-value WebSocket/stream subscriptions with automatic cleanup
    • Custom cancellation via onDisplace (abort requests, close connections) before inner is destroyed

    Universal composite transfer — combines an input and an output transfer into a single duplex interface. Automatically extracts triggerable and gate from the provided transfers (or accepts them explicitly).

    Extraction priorities:

    1. Explicit specification in config (config.triggerable / config.gate)
    2. config.input (if it has the corresponding flag)
    3. config.output (if it has the corresponding flag)
    import { createPushStoredChannelTransfer, UniversalCompositeTransfer } from 'transferum';

    const transfer = createPushStoredChannelTransfer<number>();

    const composite = new UniversalCompositeTransfer({
      input: transfer,
      output: transfer,
      owned: [transfer], // resources to destroy on destroy()
    });

    composite.push(42);
    composite.subscribe((data) => console.log(data));
    composite.pull();
    composite.trigger();

    composite.destroy(); // destroys all owned resources

    onStateChange() delegates to the internal _gate (if extracted). The subscriber receives the internal gate's GateInterface, not the composite itself. If no gate is extracted (isGate === false), onStateChange() throws an Error.

    Async methods (delegated to input/output/asyncTriggerable):

    Method Delegation Condition
    asyncPush(data) _input.asyncPush(data) isAsyncPushable === true
    asyncPull() _output.asyncPull() isAsyncPullable === true
    asyncTrigger() _asyncTriggerable.asyncTrigger() isAsyncTriggerable === true
    setAsyncFetcher(f) _input.setAsyncFetcher(f) isAsyncPollingProxy === true
    clearAsyncFetcher() _input.clearAsyncFetcher() isAsyncPollingProxy === true

    asyncTriggerable is extracted using the same priorities as triggerable/gate: explicit config → input → output.

    Async transfers provide asynchronous interfaces (asyncPush, asyncPull, asyncTrigger) for integration with asynchronous data sources (API, IndexedDB, fetch). Subscription remains synchronous in all async transfers — subscribers are notified synchronously, even if data is obtained via an async operation.

    Why separate async classes? Promise<T> and T are different programming models — mixing them in one class blurs contracts and complicates error handling. Transferum keeps sync and async as distinct transfer families (e.g., ConvertTransfer vs. AsyncConvertTransfer, ConditionTransfer vs. AsyncConditionTransfer). This increases the class count but makes each transfer's contract unambiguous: a sync transfer's push() returns void, an async transfer's asyncPush() returns Promise<void>. The type system enforces the difference — you cannot accidentally call await on a sync transfer or forget await on an async one.

    Async terminal sink — calls a callback on receiving data via asyncPush.

    Capabilities: isInput, isAsyncPushable

    Backpressure: Optional maxConcurrency, bufferSize, onBufferOverflow (see Backpressure).

    Ordered execution: Optional ordered: true — callback invocations are executed sequentially in data-arrival order, regardless of their async duration. Default: false (unordered, backward-compatible).

    Error handling: If callback() throws, onError is called. With onError provided, the exception is suppressed. Without onError — rethrown.

    import { createAsyncSinkTransfer } from 'transferum';

    const sink = createAsyncSinkTransfer<number>({
      callback: async (n) => { await fetch('/api', { body: JSON.stringify(n) }); },
      onError: (e) => console.error(e),
    });

    await sink.asyncPush(42); // → await callback(42)

    Async write adapter for AsyncInputFlowInterface (or synchronous InputFlowInterface).

    Capabilities: isInput, isAsyncPushable

    Backpressure: Optional maxConcurrency, bufferSize, onBufferOverflow (see Backpressure).

    Ordered execution: Optional ordered: true — flow.write() invocations are executed sequentially in data-arrival order, regardless of their async duration. Default: false (unordered, backward-compatible).

    import { createAsyncWriteTransfer } from 'transferum';

    const writer = createAsyncWriteTransfer<number>({ flow: asyncStorage });

    await writer.asyncPush(42); // → await flow.write(42)

    Async read adapter for AsyncOutputFlowInterface (or synchronous OutputFlowInterface).

    Capabilities: isOutput, isAsyncPullable

    import { createAsyncReadTransfer } from 'transferum';

    const reader = createAsyncReadTransfer<number>({ flow: asyncStorage });

    const value = await reader.asyncPull(); // → await flow.read()

    Async converter transfer: transforms input data via an AsyncOperator and sends the result to subscribers. If the operator returns undefined, subscribers are not notified.

    Capabilities: isInput, isOutput, isDuplex, isAsyncPushable, isSubscribable

    Backpressure: Optional maxConcurrency, bufferSize, onBufferOverflow (see Backpressure).

    Sequence Guard (active when maxConcurrency > 1): Multiple operator.apply() calls run in parallel, but results are emitted to subscribers strictly in data-arrival order. A faster result waits in an internal pending queue until all preceding operations have emitted. This prevents a stale result from overwriting a fresh one. When maxConcurrency <= 1, the guard is inactive (no overhead).

    import { createAsyncConvertTransfer, createAsyncMapOperator } from 'transferum';

    const converter = createAsyncConvertTransfer<number, string>({
      operator: createAsyncMapOperator(async (n: number) => `val_${n}`),
    });

    converter.subscribe((data) => console.log(data));
    await converter.asyncPush(42); // → "val_42"

    Transfer with asynchronous conditional filtering. The shouldAccept and shouldEmit predicates can be sync or async (return Promise<boolean> | boolean).

    Capabilities: isInput, isOutput, isDuplex, isAsyncPushable, isSubscribable

    Backpressure: Optional maxConcurrency, bufferSize, onBufferOverflow (see Backpressure).

    Sequence Guard (active when maxConcurrency > 1): Multiple shouldAccept/shouldEmit checks run in parallel, but emissions to subscribers happen strictly in data-arrival order. shouldEmit receives the operation's local data: T, not a shared state value. When maxConcurrency <= 1, the guard is inactive (no overhead).

    import { createAsyncConditionTransfer } from 'transferum';

    const condition = createAsyncConditionTransfer<number>({
      shouldAccept: async (n) => (await check(n)).valid,
      shouldEmit: (n) => n < 100,
    });

    condition.subscribe((data) => console.log(data));
    await condition.asyncPush(42); // → passes both filters → 42

    Output transfer with asynchronous internal polling. The ticker calls asyncTrigger() (fire-and-forget). The _polling flag prevents overlapping calls with a slow fetcher.

    Capabilities: isOutput, isPollingSource, isAsyncPullable, isSubscribable, isAsyncTriggerable, isGate

    import { createAsyncPollingSourceTransfer } from 'transferum';

    const polling = createAsyncPollingSourceTransfer<number>({
      fetcher: async () => (await fetch('/api/data')).json(),
      interval: 1000,
      activated: true,
    });

    polling.subscribe((data) => console.log(data));
    // Every second: async-fetch → subscriber notification

    Error handling: If fetcher() throws in asyncTrigger(), onError is called. With onError — suppressed, polling continues. Without onError (or if it throws) — rejection rethrown, ticker stops, resulting in an unhandled promise rejection (ticker calls asyncTrigger() fire-and-forget). asyncPull() rethrows to caller without affecting the ticker.

    import { createAsyncPollingSourceTransfer } from 'transferum';

    const polling = createAsyncPollingSourceTransfer<number>({
      fetcher: async () => { throw new Error('fail'); },
      interval: 1000,
      activated: true,
      onError: (e, source) => console.error(e, source.active),
    });

    Duplex transfer with async polling that receives its fetcher from the previous node. setAsyncFetcher() is set via linkTransfers (async strategies 5–7).

    Capabilities: isInput, isOutput, isDuplex, isAsyncPollingProxy, isPollingSource, isAsyncPullable, isSubscribable, isAsyncTriggerable, isGate

    import { createAsyncPollingProxyTransfer } from 'transferum';

    const poller = createAsyncPollingProxyTransfer<number>({
      interval: 1000,
      activated: false,
    });

    // asyncFetcher is set via linkTransfers or manually:
    poller.setAsyncFetcher(async () => await asyncSource.asyncPull());

    poller.activate(); // start polling

    Error handling: Same as AsyncPollingSourceTransfer — onError suppresses fetcher errors in asyncTrigger() and asyncPull(). Without onError (or if it throws) — asyncTrigger() rethrows, ticker stops, unhandled promise rejection. asyncPull() rethrows to caller without affecting the ticker.

    Output transfer with async polling from AsyncOutputFlowInterface. Similar to AsyncPollingSourceTransfer, but the source is an interface with async read() instead of AsyncDataFetcher.

    Capabilities: isOutput, isPollingSource, isAsyncPullable, isSubscribable, isAsyncTriggerable, isGate

    import { createAsyncPollingFlowTransfer } from 'transferum';

    const polling = createAsyncPollingFlowTransfer<number>({
      flow: asyncStorage,
      interval: 1000,
      activated: true,
    });

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

    Error handling: If flow.read() throws in asyncTrigger(), onError is called. With onError — suppressed, polling continues. Without onError (or if it throws) — rejection rethrown, ticker stops, unhandled promise rejection. asyncPull() rethrows to caller without affecting the ticker.

    Reactive channel with async fallback polling on idle. push() is synchronous, but the fetcher is asynchronous — asyncTrigger() awaits _doPoll() (fetch + notify), asyncPull() awaits the fetcher directly. The _polling flag prevents overlapping. The ticker uses _doPoll() — fire-and-forget.

    Capabilities: isInput, isOutput, isDuplex, isPushable, isSubscribable, isPollingSource, isAsyncPullable, isAsyncTriggerable, isGate

    import { createAsyncIdlePollingTransfer } from 'transferum';

    const channel = createAsyncIdlePollingTransfer<number>({
      fetcher: async () => (await fetch('/api/latest')).json(),
      timeout: 5000,   // 5 seconds without push → start polling
      interval: 1000,  // poll fetcher every second
      activated: true,
    });

    channel.subscribe((data) => console.log(data));
    channel.push(42);  // → subscribers notified synchronously, idle timer reset
    // after 5 seconds without push → async-polling every 1 second

    Error handling: If fetcher() throws during polling, onError is called. With onError — suppressed, polling continues. Without onError (or if it throws) — rejection rethrown, polling stops, unhandled promise rejection (ticker calls _doPoll() fire-and-forget). asyncPull() rethrows to caller without affecting polling.

    Channel with value retention, external management, and async interface. setup/emit/subscribe are synchronous (like in StoredChannelTransfer), but asyncPull()/asyncTrigger() are async for integration with async pipelines.

    Capabilities: isOutput, isSubscribable, isAsyncPullable, isAsyncTriggerable

    Error handling: Same as ChannelTransfer — setup() errors always rethrown. onError covers emit() failures (via asyncTrigger()). onDestroyError covers destroy() failures.

    import { createAsyncStoredChannelTransfer } from 'transferum';

    let emit: (data: number) => void;

    const channel = createAsyncStoredChannelTransfer<number>({
      setup: (e) => { emit = e; },
      destroy: () => {},
      initialValue: 0,
    });

    channel.subscribe((data) => console.log(data));
    emit(42); // → 42 (asyncTrigger — fire-and-forget)

    const value = await channel.asyncPull(); // 42

    Four async transfers — AsyncSinkTransfer, AsyncWriteTransfer, AsyncConvertTransfer, and AsyncConditionTransfer — support optional backpressure via the shared BackpressureConfig<T>:

    Option Type Default Description
    maxConcurrency number Infinity Maximum number of concurrent async operations. Excess data is queued in the buffer.
    bufferSize number Infinity Maximum items held in the buffer while all concurrency slots are occupied.
    onBufferOverflow DataHandler<T> undefined Called when both concurrency and buffer are full. If omitted, excess data is silently dropped.

    Mechanics:

    1. asyncPush(data) — if _activeCount < maxConcurrency, data is processed immediately.
    2. If at capacity and _buffer.length < bufferSize — data is queued in _buffer.
    3. If both are full — onBufferOverflow?.(data) is called (or data is silently dropped if no handler).
    4. When an active operation completes (success or error), _dequeue() shifts the next buffered item and processes it.
    5. destroy() clears the buffer — queued items are discarded.

    All defaults are Infinity, so existing code is fully backward-compatible — without backpressure config, all four transfers behave exactly as before (unlimited parallel processing, no buffering).

    Ordered execution:

    • AsyncSinkTransfer and AsyncWriteTransfer: optional ordered: true — callback/write invocations are executed sequentially in data-arrival order via an internal OrderedExecutor. Default: false (unordered, backward-compatible).
    • AsyncConvertTransfer and AsyncConditionTransfer: automatic Sequence Guard when maxConcurrency > 1 — results are emitted to subscribers strictly in data-arrival order, even though operations run in parallel. No config needed. When maxConcurrency <= 1, the guard is inactive (zero overhead).
    import { createAsyncWriteTransfer } from 'transferum';

    const writer = createAsyncWriteTransfer<number>({
      flow: asyncStorage,
      maxConcurrency: 3,   // at most 3 concurrent flow.write() calls
      bufferSize: 10,      // queue up to 10 items
      onBufferOverflow: (data) => console.warn('Dropped:', data),
    });

    // Fire-and-forget — excess pushes are buffered or dropped
    writer.asyncPush(1);
    writer.asyncPush(2);
    // ...

    Operators implement OperatorInterface<TInput, TOutput> with the method apply(data: TInput): TOutput.

    Operators are pure transforms — transfers carry the contract. In RxJS, an operator transforms Observable<A> into Observable<B> — the container type stays the same. In Transferum, operators are stateless data transformers used inside transfers (ConvertTransfer, TransformBridge). The transfer's capability flags (pushable, subscribable, gate, etc.) are preserved through the transformation — a ConvertTransfer is still pushable and subscribable, just with different input/output types. This separation means operators focus purely on data transformation, while transfers own the behavioral contract.

    Operator Input Output Behavior Composition Use case
    TransparentOperator<T> T T Returns data unchanged (identity) Yes Stub, pipeline testing without transformation
    MapOperator<TIn, TOut> TIn TOut Applies a mapper function to data Yes Type or value transformation
    FilterOperator<T> T[] T[] Keeps array elements matching a predicate Yes Array element filtering
    ReducerOperator<T> T[] T | undefined Reduces an array to a single value (undefined for empty without default) Yes Aggregation (sum, product, min/max)
    GuardOperator<T> T T | undefined Passes data if predicate is true, otherwise undefined Yes Validation, single-value filtering
    PipelineOperator<TIn, TOut> TIn TOut Sequentially applies a chain of operators Composition result Complex multi-stage transformations
    AsyncMapOperator<TIn, TOut> TIn Promise<TOut> Asynchronously applies a mapper (sync or async) Yes (via AsyncPipeline) Async type or value transformation
    AsyncGuardOperator<T> T Promise<T | undefined> Asynchronously passes data if predicate is true Yes (via AsyncPipeline) Async validation, filtering
    AsyncPipelineOperator<TIn, TOut> TIn Promise<TOut> Sequentially applies a chain of sync/async operators via await Composition result Complex async multi-stage transformations

    Composition — the ability to use an operator as a step in PipelineOperator (via OperatorPipelineBuilder) or as an operator in ConvertTransfer / TransformBridge. All operators support composition.

    Comparison by data types:

    Characteristic Scalar operators Array operators
    Operators TransparentOperator, MapOperator, GuardOperator FilterOperator, ReducerOperator
    Input Single value (T) Array (T[])
    Output Single value or undefined Array, scalar, or undefined
    GuardOperator Returns T | undefined — blocks the flow on false —
    FilterOperator — Returns T[] — subset of elements
    ReducerOperator — Returns T | undefined — array aggregate 
    import { createPipelineOperator, createMapOperator, createGuardOperator } from 'transferum';

    const op = createPipelineOperator<number, string>([
      createMapOperator((n: number) => n * 2),
      createMapOperator((n: number) => n.toString()),
      createGuardOperator((s: string) => s.length > 0),
    ]);

    console.log(op.apply(21)); // "42"

    Async operators implement AsyncOperatorInterface<TInput, TOutput> with the method apply(data: TInput): Promise<TOutput>. Mappers and predicates can be synchronous or asynchronous (return Promise).

    import { createAsyncPipelineOperator, createAsyncMapOperator, createMapOperator, createAsyncGuardOperator } from 'transferum';

    const op = createAsyncPipelineOperator<number, string>([
      createAsyncMapOperator(async (n: number) => n * 2),
      createMapOperator((n: number) => n.toString()),      // sync operator in async chain
      createAsyncGuardOperator(async (s: string) => s.length > 0),
    ]);

    console.log(await op.apply(21)); // "42"

    AsyncPipelineOperator accepts both OperatorInterface and AsyncOperatorInterface — each step is executed via await. A sync operator is unwrapped as a no-op.

    AsyncPipelineOperator (like the sync PipelineOperator) does not stop the chain on undefined from a guard — it passes undefined further.

    Storages implement StorageInterface<TInput, TOutput> (write/read/clear/reset/size).

    Storage Structure Read order maxLength size reset() Purpose
    LatestStorage<T> Single value — (last written) — 0 or 1 Restores defaultValue Last-value cache
    QueueStorage<T> Array (FIFO) First written → first read ✓ (evicts oldest) Element count Clears (clear()) FIFO buffer with limit
    StackStorage<T> Array (LIFO) Last written → first read ✓ (evicts from bottom) Element count Clears (clear()) LIFO stack with limit

    Behavior comparison:

    Characteristic LatestStorage QueueStorage StackStorage
    Stores multiple values — ✓ ✓
    Read order Last FIFO (first in — first out) LIFO (last in — first out)
    write() overwrites ✓ (always) ✓ (only at maxLength) ✓ (only at maxLength)
    read() clears value — (available repeatedly) ✓ (extracts and removes) ✓ (extracts and removes)
    defaultValue in constructor ✓ — —
    reset() Restores defaultValue Clears Clears
    maxLength — ✓ (removes oldest) ✓ (removes oldest from bottom)
    size after clear() 0 0 0

    Choosing a storage:

    Scenario Recommendation
    Caching the last state LatestStorage
    Buffering data in arrival order QueueStorage
    Reverse processing (last in — first processed) StackStorage
    Limiting stored data volume QueueStorage or StackStorage with maxLength
    Source for PollingFlowTransfer Any (via ReadTransfer)
    Sink for WriteTransfer Any (via WriteTransfer) 
    import { createQueueStorage } from 'transferum';

    const queue = createQueueStorage<number>(3);
    queue.write(1);
    queue.write(2);
    queue.write(3);
    queue.write(4); // 1 evicted

    console.log(queue.read()); // 2
    console.log(queue.size);   // 2

    Tickers implement TickerInterface and provide periodic callback invocation with a configurable interval. Two implementations for different environments:

    Ticker Based on Leading edge Environment
    RAFTicker requestAnimationFrame ✓ (first frame) Browser / SSR (setTimeout fallback)
    IntervalTicker setInterval ✓ (setTimeout(fn, 0)) Node.js / tests (fake timers)

    Interface:

    interface TickerInterface {
      readonly interval: number;
      readonly active: boolean;
      start(): void;
      stop(): void;
      restart(): void;
      toggle(): boolean;
      updateInterval(delay: number): void;
    }

    Usage example:

    import type { TickerFactory } from 'transferum';
    import { RAFTicker, IntervalTicker } from 'transferum';

    // Browser ticker (default)
    const rafTicker = new RAFTicker({ callback: () => console.log('tick'), interval: 1000 });
    rafTicker.start();

    // Server/test ticker
    const intervalTicker = new IntervalTicker({ callback: () => console.log('tick'), interval: 1000 });
    intervalTicker.start();

    // Via factory (for passing to polling transfers)
    const tickerFactory: TickerFactory = (config) => new RAFTicker(config);

    Leading edge: Both tickers invoke the callback immediately on start() (or in the next micro-tick for IntervalTicker when interval > 0). When interval === 0, IntervalTicker starts setInterval(fn, 0) without delay.

    Safe stop() inside callback: RAFTicker recalculates _startTime before calling callback(), so the callback can safely call stop() synchronously — the frame will not be rescheduled.

    Subscription management. Created via SubscriptionManager.subscribe(), not directly.

    // transfer is any SubscribableTransferInterface
    const subscriber = transfer.subscribe(handler);

    subscriber.onUnsubscribe((s) => console.log('unsubscribed'));
    subscriber.unsubscribe(); // → "unsubscribed"

    Manages a set of subscribers. sendState() notifies all subscribers with the current value (ignores undefined).

    A wrapper around a value:

    • value — current value
    • pop() — extract with cleanup
    • clear() — clear without extraction

    Adapts SubscriberInterface → DisposableInterface. Used in builders to manage subscriptions via destroy().

    Subscription manager for object state changes. Reuses ProxyReference and SubscriptionManager.

    The value (usually the owning object itself) is set once in the constructor and never becomes undefined, so every notify() is guaranteed to notify all subscribers with that value.

    Used to implement GateInterface.onStateChange() in all gate transfers and bridges.

    import type { GateInterface } from 'transferum';
    import { StateSubscriptionManager, createGateTransfer } from 'transferum';

    const gate = createGateTransfer({
      activated: true;
    })

    const manager = new StateSubscriptionManager<GateInterface>(gate);

    const subscriber = manager.subscribe((g) => {
      console.log(`State changed: active=${g.active}`);
    });

    gate.activate();
    manager.notify(); // → "State changed: active=true"

    subscriber.unsubscribe(); // unsubscribe

    manager.destroy(); // unsubscribes all remaining subscribers

    Methods:

    Method Description
    subscribe(handler) Registers a handler, returns SubscriberInterface
    notify() Notifies all active subscribers with the value from the constructor
    destroy() Unsubscribes all active subscribers

    Bridges implement BridgeInterface (active/activate/deactivate/toggle/destroy) and connect data flows with gate control.

    Bridge is a first-class entity, not a helper method. In most libraries, connecting two nodes is a method call: source.connect(target) or source.pipe(target). This forces every node to know about every other node type — coupling grows quadratically. Transferum makes connection a separate entity: Transfer is the node (defines behavior), Bridge is the edge (defines interaction). A bridge inspects capability flags, not class names — the same principle as separating nodes and edges in a graph, or TCP from applications in a network stack.

    Bridge Connects Transformation Intermediate transfer Gate Owned Purpose
    PassBridge<T> Output → Input — — ✓ — Simple bridge with flow control
    TransformBridge<TIn, TOut> Output → Input ✓ (via Operator) ConvertTransfer (internal) ✓ — Bridge with data type transformation
    TransferBridge<TIn, TOut> Output → Input — DuplexTransfer (external, opt. middleOwned) ✓ ✓ Bridge with intermediate duplex transfer
    AsyncTransformBridge<TIn, TOut> Output → Input ✓ (via AsyncOperator) AsyncConvertTransfer (internal) ✓ — Bridge with async type transformation
    BridgeAggregator Group of bridges — — ✓ (all at once) ✓ Synchronous group control of bridges
    BridgeSelector<TMap> One from a bridge map — — ✓ (only selected) ✓ Select one active bridge
    BridgeMultiSelector<TMap> Several from a bridge map — — ✓ (selected) ✓ Select multiple active bridges

    Gate — all bridges have an internal GateTransfer for flow control. activate() / deactivate() / toggle() delegate to the gate. Owned — the owned parameter controls whether nested bridges are destroyed on destroy(). Link Strategy — PassBridge, TransformBridge, TransferBridge, and AsyncTransformBridge accept an optional linkStrategy?: LinkStrategyInterface in their config. When provided, all internal links (source → gate → converter → target) are created via linkStrategy.link() instead of linkTransfers(). This allows custom link strategies to control bridge-internal wiring consistently with CompositeTransferBuilder. See Linking. onStateChange() — all bridges support state change subscription via onStateChange(). In PassBridge, TransformBridge, TransferBridge the notification fires on activate() / deactivate() / toggle(). In BridgeAggregator — on direct state changes (does not listen to children). In BridgeSelector — on _active or _selectedKey changes (including select()). In BridgeMultiSelector — on _active or _selectedKeys changes (including select(), check(), uncheck()). When syncWithChildren is enabled, BridgeSelector and BridgeMultiSelector also fire onStateChange() when a child bridge's state changes externally — the selector reacts by updating its selection and notifying its own subscribers.

    Comparison by flow structure:

    Characteristic PassBridge TransformBridge TransferBridge
    Connects source → target source → target source → target
    Intermediate layer — ConvertTransfer External DuplexTransfer
    Data transformation — ✓ (Operator) Depends on middle
    Data types T → T TIn → TOut TIn → TOut
    Middle management — Internal (created by bridge) External (middleOwned controls destroy)
    Link chain source → gate → target source → gate → converter → target source → gate → middle → target

    Comparison of selectors and aggregator:

    Characteristic BridgeAggregator BridgeSelector BridgeMultiSelector
    Active bridges All simultaneously One (selected) Several (selected)
    active true if ALL bridges are active Controlled by flag Controlled by flag
    Selection — select(key) select(keys[]), check(key), uncheck(key)
    Switching activate() / deactivate() all select(key) switches check() / uncheck() add/remove
    Extraction — selectedKey, selectedBridge selectedKeys, selectedBridges
    owned Controls destroy of all bridges Controls destroy of all bridges Controls destroy of all bridges
    syncWithChildren — ✓ (optional, default false) ✓ (optional, default false)
    toggle() Activates/deactivates all Toggles common flag Toggles common flag

    Choosing a bridge:

    Scenario Recommendation
    Simple passthrough with on/off capability PassBridge
    Passthrough with data type transformation TransformBridge
    Passthrough through an intermediate transfer (filter, converter) TransferBridge
    Synchronous group control of bridges BridgeAggregator
    Switching between multiple routes BridgeSelector
    Simultaneous activation of multiple routes BridgeMultiSelector 
    import { createPushStoredChannelTransfer, createSinkTransfer, createPassBridge } from 'transferum';

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

    const bridge = createPassBridge<number>({
      source,
      target,
      activated: true,
    });

    source.push(42); // → callback called

    bridge.deactivate();
    source.push(100); // ignored

    bridge.destroy(); // breaks all links

    import { createBridgeSelector, createPassBridge } from 'transferum';

    const bridges = {
      fast: createPassBridge({ source, target: target1, activated: false }),
      slow: createPassBridge({ source, target: target2, activated: false }),
    };

    const selector = createBridgeSelector({
      bridges,
      initialKey: 'fast',
      activated: true,
      owned: false,
    });

    selector.select('slow'); // switch to the second bridge

    syncWithChildren (optional, default false): when enabled, the selector subscribes to onStateChange() of all child bridges and reacts to their external state changes:

    • A child bridge activated externally → the selector switches selection to that bridge (select()).
    • The selected bridge deactivated externally → the selector deactivates itself (deactivate()).
    • State changes of non-selected bridges (when deactivated) or already-selected bridges (when activated) are ignored.

    This enables bidirectional gate synchronization: the selector controls its children, and the children can control the selector. An internal _syncing guard prevents feedback loops — when the selector itself activates/deactivates/selects, child state-change notifications are suppressed.

    import { createBridgeSelector } from 'transferum';

    const selector = createBridgeSelector({
      bridges,
      initialKey: 'fast',
      activated: true,
      syncWithChildren: true,
      owned: false,
    });

    // External activation of 'slow' bridge → selector automatically switches to 'slow'
    bridges.slow.activate();
    console.log(selector.selectedKey); // 'slow'

    import { createBridgeMultiSelector } from 'transferum';

    const selector = createBridgeMultiSelector({
      bridges,
      initialKeys: ['fast'],
      activated: true,
      owned: false,
    });

    selector.check('slow');   // adds bridge to active
    selector.uncheck('fast'); // removes bridge from active

    syncWithChildren (optional, default false): when enabled, the selector subscribes to onStateChange() of all child bridges and reacts to their external state changes:

    • A child bridge activated externally → the selector adds it to the selection (check()).
    • A selected child bridge deactivated externally → the selector removes it from the selection (uncheck()).
    • State changes of already-selected bridges (when activated) or non-selected bridges (when deactivated) are ignored.

    This enables bidirectional gate synchronization: the selector controls its children, and the children can control the selector. An internal _syncing guard prevents feedback loops — when the selector itself activates/deactivates/selects/checks/unchecks, child state-change notifications are suppressed.

    import { createBridgeMultiSelector } from 'transferum';

    const selector = createBridgeMultiSelector({
      bridges,
      initialKeys: ['fast'],
      activated: true,
      syncWithChildren: true,
      owned: false,
    });

    // External activation of 'slow' bridge → selector automatically adds it to selection
    bridges.slow.activate();
    console.log(selector.selectedKeys); // ['fast', 'slow']

    // External deactivation of 'fast' bridge → selector automatically removes it
    bridges.fast.deactivate();
    console.log(selector.selectedKeys); // ['slow']

    Bridge with asynchronous data type transformation via AsyncOperator. Gate and subscription remain synchronous. AsyncConvertTransfer accepts data via asyncPush (the gate→converter link uses the async linkTransfers strategy), transforms via await operator.apply(), and notifies subscribers synchronously.

    Flow structure: source → gate → asyncConverter → target

    import { createAsyncTransformBridge, createAsyncMapOperator } from 'transferum';

    const bridge = createAsyncTransformBridge<number, string>({
      source,
      target,
      operator: createAsyncMapOperator(async (n: number) => `val_${n}`),
      activated: true,
      onError: (e) => console.error(e),
    });

    source.push(42); // → async transformation → "val_42" at target's subscribers

    Builders provide a fluent API for assembling transfer chains with automatic linking via linkTransfers. A custom LinkStrategyInterface can be injected into start() to override linking behavior for the entire chain.

    CompositeTransferBuilder is the unified, type-safe builder that replaces InputPipelineBuilder, OutputPipelineBuilder, DuplexPipelineBuilder, and all async variants. A single builder covers all pipeline directions.

    Pipeline structure: OutputTransfer [→ DuplexTransfer → …] → InputTransfer

    import {
      CompositeTransferBuilder,
      createPushStoredChannelTransfer,
      createConditionTransfer,
      createSinkTransfer,
      createPollingSourceTransfer,
      createConvertTransfer,
      createMapOperator,
    } from 'transferum';

    // Input pipeline: push → condition → sink
    const input = CompositeTransferBuilder
      .start(createPushStoredChannelTransfer<number>())
      .to(createConditionTransfer<number>({ shouldAccept: x => x > 0 }))
      .finish(createSinkTransfer<number>({ callback: console.log }), { owned: true });

    input.push(42); // → 42

    // Output pipeline: polling → convert → stored channel
    const output = CompositeTransferBuilder
      .start(createPollingSourceTransfer<number>({ fetcher: () => 42, interval: 1000, activated: true }))
      .to(createConvertTransfer<number, string>({ operator: createMapOperator(n => n.toString()) }))
      .finish(createPushStoredChannelTransfer<string>());

    output.subscribe(data => console.log(data)); // → "42" every second

    // Full-duplex pipeline: push → condition → stored channel (push + subscribe + pull)
    const duplex = CompositeTransferBuilder
      .start(createPushStoredChannelTransfer<number>())
      .to(createConditionTransfer<number>({ shouldAccept: x => x > 0 }))
      .finish(createPushStoredChannelTransfer<number>(), { owned: true });

    duplex.push(42);
    duplex.subscribe(data => console.log(data)); // → 42
    duplex.pull(); // → 42
    duplex.destroy(); // destroys owned resources

    Auto-capability inference:

    • Input flags (Pushable, PollingProxy, AsyncPushable, AsyncPollingProxy) are extracted from the start transfer.
    • Output flags (Pullable, Subscribable, AsyncPullable) are extracted from the finish transfer.
    • Triggerable, AsyncTriggerable, and Gate are inferred from explicit options or auto-extracted from the chain.

    start(startTransfer, options?):

    CompositeTransferBuilder.start(startTransfer, options?)
    // options?: { linkStrategy?: LinkStrategyInterface }
    //          custom link strategy for all to()/finish() calls
    //          if omitted, defaults to DefaultLinkStrategy

    to(transfer, options):

    to(nextTransfer, {
      owned?: boolean,                    // destroy nextTransfer on composite destroy()
      onLinkError?: ErrorHandler,         // async linking error handler
    })

    finish(lastTransfer, options):

    finish(lastTransfer, {
      triggerable?: TriggerableInterface,           // explicit trigger (priority over auto-extraction)
      asyncTriggerable?: AsyncTriggerableInterface, // explicit async trigger
      gate?: GateInterface,                         // explicit gate (priority over auto-extraction)
      owned?: boolean,                              // whether to destroy lastTransfer on destroy()
      onLinkError?: ErrorHandler,                   // async linking error handler
    })

    Unified sync + async: The onLinkError option in to() and finish() enables async error handling across the entire chain, eliminating the need for separate async builder variants.

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

    const pipeline = CompositeTransferBuilder
      .start(createPushStoredChannelTransfer<number>())
      .to(createAsyncConvertTransfer<number, string>({
        operator: createAsyncMapOperator(async (n) => n.toString()),
      }), { onLinkError: (e) => console.error(e) })
      .finish(createPushStoredChannelTransfer<string>(), { owned: true, onLinkError: (e) => console.error(e) });

    pipeline.push(42);
    pipeline.subscribe((data) => console.log(data)); // → "42"

    Custom link strategy: Pass a LinkStrategyInterface to start() to override linking for the entire chain. See Linking.

    import {
      CompositeTransferBuilder,
      DefaultLinkStrategy,
      createPushStoredChannelTransfer,
      createConditionTransfer,
      createSinkTransfer,
    } from 'transferum';

    const linkStrategy = new DefaultLinkStrategy();

    const pipeline = CompositeTransferBuilder
      .start(createPushStoredChannelTransfer<number>(), { linkStrategy })
      .to(createConditionTransfer<number>({ shouldAccept: x => x > 0 }))
      .finish(createSinkTransfer<number>({ callback: console.log }));

    pipeline.push(42); // → 42
    • owned: true in to(transfer, { owned?: boolean, ... }) — the intermediate transfer is destroyed on composite destroy().
    • owned: true in finish(lastTransfer, { owned?: boolean, ... }) — the final transfer is destroyed on composite destroy().
    • owned: false (default) — the transfer is not destroyed automatically.

    @deprecated The following builders are deprecated and will be removed in the next major release. Use CompositeTransferBuilder instead.

    Builder Status
    InputPipelineBuilder @deprecated
    OutputPipelineBuilder @deprecated
    DuplexPipelineBuilder @deprecated
    AsyncInputPipelineBuilder @deprecated
    AsyncOutputPipelineBuilder @deprecated
    AsyncDuplexPipelineBuilder @deprecated

    OperatorPipelineBuilder and AsyncOperatorPipelineBuilder work with OperatorInterface / AsyncOperatorInterface (not TransferInterface) and remain non-deprecated.

    Builds a chain of operators with type checking at each step.

    import { OperatorPipelineBuilder, createMapOperator, createGuardOperator } from 'transferum';

    const operator = OperatorPipelineBuilder
      .create()
      .add(createMapOperator<number, number>((n) => n * 2))
      .add(createMapOperator<number, string>((n) => n.toString()))
      .add(createGuardOperator<string>((s) => s.length > 0))
      .build();

    console.log(operator.apply(21)); // "42"

    Accepts both sync and async operators, build() returns AsyncPipelineOperator.

    import { AsyncOperatorPipelineBuilder, createMapOperator, createAsyncMapOperator } from 'transferum';

    const operator = AsyncOperatorPipelineBuilder
      .create()
      .add(createMapOperator<number, number>((n) => n * 2))                   // sync operator
      .add(createAsyncMapOperator<number, string>(async (n) => n.toString())) // async operator
      .build();

    console.log(await operator.apply(21)); // "42"

    Factory functions create* are convenient wrappers over constructors. The return type is computed via Transfer<TIn, TOut, [Features...]>, ensuring a precise interface.

    import { createPushChannelTransfer, createPushStoredChannelTransfer } from 'transferum';

    const channel = createPushChannelTransfer<number>();
    // type: Transfer<number, [Pushable, Subscribable]>
    // available methods: push(), subscribe(), destroy()

    const stored = createPushStoredChannelTransfer<number>({ initialValue: 0 });
    // type: Transfer<number, [Pushable, Pullable, Subscribable, Triggerable]>
    // available methods: push(), pull(), subscribe(), trigger(), destroy()

    Full list of factories:

    Category Factories
    Channels createPushChannelTransfer, createDelayedPushChannelTransfer, createDebounceTransfer, createThrottleTransfer, createPushStoredChannelTransfer
    Buffers createBufferTransfer, createManualBufferTransfer, createManualFlowTransfer
    Gate createGateTransfer
    Aggregation createMergeTransfer, createSplitTransfer
    Polling createPollingSourceTransfer, createPollingProxyTransfer, createPollingFlowTransfer, createIdlePollingTransfer
    Externally-managed channels createChannelTransfer, createStoredChannelTransfer
    Sink / Flow createSinkTransfer, createWriteTransfer, createReadTransfer
    Transformation createConvertTransfer, createConditionTransfer, createDisplaceTransfer
    Bridges createPassBridge, createTransformBridge, createTransferBridge, createBridgeAggregator, createBridgeSelector, createBridgeMultiSelector
    Linking createDefaultLinkStrategy
    Operators createTransparentOperator, createMapOperator, createFilterOperator, createReducerOperator, createGuardOperator, createPipelineOperator
    Storages createLatestStorage, createQueueStorage, createStackStorage
    Async adapters createAsyncSinkTransfer, createAsyncWriteTransfer, createAsyncReadTransfer
    Async transformation createAsyncConvertTransfer, createAsyncConditionTransfer
    Async polling createAsyncPollingSourceTransfer, createAsyncPollingProxyTransfer, createAsyncPollingFlowTransfer, createAsyncIdlePollingTransfer
    Async external sources createAsyncStoredChannelTransfer
    Async bridges createAsyncTransformBridge
    Async operators createAsyncMapOperator, createAsyncGuardOperator, createAsyncPipelineOperator 
    function linkTransfers<T, RTransfer extends InputTransfer<T>>(
      lhs: OutputTransfer<T>,
      rhs: RTransfer,
      options?: LinkConfig<RTransfer>,
    ): SubscriberInterface

    Links an output transfer (LHS) to an input transfer (RHS). Returns SubscriberInterface for breaking the link. The strategy is determined by capability flags (see Linking Transfers). options.onError is used to intercept rejections in the async subscribable → asyncPushable strategy — invoked as onError(error, target) via handleError(). Without onError, rejections are rethrown by handleError() (unhandled promise rejection); the source's subscription remains active.

    Internally, linkTransfers dispatches to one of seven exported strategy functions based on capability flags. You can call these directly for the same result:

    Strategy LHS RHS Behavior
    linkSubscribableToPushable Subscribable Pushable Reactive subscription: LHS notifies → RHS accepts
    linkPullableToPollingProxy Pullable PollingProxy Active polling: RHS pulls via setFetcher
    linkSubscribableToPollingProxy Subscribable PollingProxy Subscription + last-value buffering for the poller
    linkSubscribableToAsyncPushable Subscribable AsyncPushable Subscription + asyncPush with .catch() (no ordering)
    linkAsyncPullableToAsyncPollingProxy AsyncPullable AsyncPollingProxy Active async polling: RHS pulls via setAsyncFetcher
    linkPullableToAsyncPollingProxy Pullable AsyncPollingProxy Sync-pull wrapped in an async fetcher
    linkSubscribableToAsyncPollingProxy Subscribable AsyncPollingProxy Subscription + buffer + async fetcher

    Error helpers for unsupported combinations:

    Function When
    throwLinkAsyncPullableToPollingProxyError AsyncPullable → sync PollingProxy (cannot await)
    throwLinkPullableToPushableError Pullable/AsyncPullable → Pushable/AsyncPushable (needs a Bridge or Triggerable adapter)
    throwLinkUnsupportedError Any other incompatible combination 
    function handleError<TSource>(error: unknown, source: TSource, onError?: ErrorHandler<TSource>): void;

    Universal error handler:

    • If onError is provided — invokes it with (error, source) and suppresses the exception.
    • If onError is not provided — rethrows the exception.
    • Non-Error values are converted to Error (via String(error)).

    Type guards for narrowing CommunicationContractInterface to a specific branded capability type. Each guard checks a boolean capability flag and, when true, narrows the TypeScript type so that the corresponding methods are available without casts.

    Guard Checks flag Narrows to Methods unlocked
    isPushable<T> isPushable Pushable<T> push(data: T)
    isPullable<T> isPullable Pullable<T> pull(): T | undefined
    isSubscribable<T> isSubscribable Subscribable<T> subscribe(handler): SubscriberInterface
    isPollingProxy<T> isPollingProxy PollingProxy<T> setFetcher(), clearFetcher()
    isTriggerable isTriggerable Triggerable trigger()
    isGate isGate Gate activate(), deactivate(), toggle(), onStateChange()
    isAsyncPushable<T> isAsyncPushable AsyncPushable<T> asyncPush(data: T): Promise<void>
    isAsyncPullable<T> isAsyncPullable AsyncPullable<T> asyncPull(): Promise<T | undefined>
    isAsyncPollingProxy<T> isAsyncPollingProxy AsyncPollingProxy<T> setAsyncFetcher(), clearAsyncFetcher()
    isAsyncTriggerable isAsyncTriggerable AsyncTriggerable asyncTrigger(): Promise<void> 
    import type { CommunicationContractInterface } from "transferum";
    import { isPushable, isSubscribable, createPushChannelTransfer } from 'transferum';

    const transfer: CommunicationContractInterface = createPushChannelTransfer<number>();

    if (isSubscribable(transfer) && isPushable(transfer)) {
      // TypeScript knows: transfer.subscribe() and transfer.push() exist
      transfer.subscribe((data) => console.log(data));
      transfer.push(42);
    }

    Internally, DefaultLinkStrategy.link() uses these guards instead of as-casts to narrow transfer types before dispatching to each linking strategy — making the linking code type-safe without runtime overhead.

    Key types are defined in types.ts:

    Type Description
    Transfer<TInOrAll, TOutOrFeatures, TFeatures?> Computed transfer type from a list of capabilities
    GateInterface Flow control: active, activate(), deactivate(), toggle(), onStateChange()
    SubscriberInterface Subscription management: active, unsubscribe(), onUnsubscribe(), offUnsubscribe()
    DisposableInterface Resource cleanup: destroy()
    LinkStrategyInterface Facade for linking transfers: link()
    InputTransfer<T> PushableTransferInterface | PollingProxyTransferInterface | GateTransferInterface
    OutputTransfer<T> PullableTransferInterface | SubscribableTransferInterface | GateTransferInterface
    DuplexTransfer<TIn, TOut> InputTransfer<TIn> & OutputTransfer<TOut>
    CompositeInputTransfer Composite input transfer (from a builder) [deprecated: use CompositeTransfer]
    CompositeOutputTransfer Composite output transfer (from a builder) [deprecated: use CompositeTransfer]
    CompositeDuplexTransfer Composite duplex transfer (from a builder) [deprecated: use CompositeTransfer]
    CompositeTransfer<TInput, TOutput, TStart, TFinish, ...> Composite transfer type with computed capability flags from start and finish transfers (new unified type)
    First<T> / Last<T> First/last element of a tuple
    InputTransferDataType<T> Extracts the data type from InputTransfer
    OutputTransferDataType<T> Extracts the data type from OutputTransfer
    AsyncDataHandler<T> Data handler: (data: T) => Promise<void> | void
    ErrorHandler<TSource> Error handler: (e: Error, source: TSource) => void
    AsyncDataFetcher<T> Data fetcher function: () => Promise<T | undefined>
    AsyncPushable<T> AsyncPushableInterface<T> & { readonly isAsyncPushable: true }
    AsyncPullable<T> AsyncPullableInterface<T> & { readonly isAsyncPullable: true }
    AsyncTriggerable AsyncTriggerableInterface & { readonly isAsyncTriggerable: true }
    AsyncPollingProxy<T> AsyncPollingProxyInterface<T> & { readonly isAsyncPollingProxy: true }
    AsyncOperatorInterface<TInput, TOutput> Async operator: apply(data: TInput): Promise<TOutput>
    AsyncInputFlowInterface<T> Async write: write(data: T): Promise<void>
    AsyncOutputFlowInterface<T> Async read: read(): Promise<T | undefined>
    AsyncIOFlowInterface<TInput, TOutput> AsyncInputFlowInterface<TInput> & AsyncOutputFlowInterface<TOutput>
    AsyncStorageInterface<TInput, TOutput> AsyncIOFlowInterface + size, clear(), reset() (async)

    Capability-derived types: InputTransfer<T>, OutputTransfer<T>, DuplexTransfer<TIn, TOut>, and Transfer<…> are not hand-written unions — they are computed from capability flag interfaces (where each flag is narrowed to true). Branded types like Pushable<T>, Subscribable<T>, Gate add a literal true brand to the corresponding flag. Builders use these types to enforce capability compatibility at compile time. See Capability Flags System.

    Configs are defined in configs.ts. All configs are types (not classes), passed to transfer and bridge constructors.

    Config For Required fields
    GateTransferConfig GateTransfer activated
    DelayedPushChannelTransferConfig<T> DelayedPushChannelTransfer delay
    DebounceTransferConfig DebounceTransfer delay
    ThrottleTransferConfig ThrottleTransfer interval
    MergeTransferConfig<T> MergeTransfer sources
    SplitTransferConfig<T> SplitTransfer targets
    PollingSourceTransferConfig<T> PollingSourceTransfer fetcher, interval, activated
    PollingProxyTransferConfig PollingProxyTransfer interval, activated
    PollingFlowTransferConfig<T> PollingFlowTransfer flow, interval, activated
    IdlePollingTransferConfig<T> IdlePollingTransfer fetcher, timeout, interval, activated
    ChannelTransferConfig<T> ChannelTransfer setup, destroy, onError?, onDestroyError?
    StoredChannelTransferConfig<T> StoredChannelTransfer setup, destroy, onError?, onDestroyError?
    SinkTransferConfig<T> SinkTransfer callback, onError?
    WriteTransferConfig<T> WriteTransfer flow
    ReadTransferConfig<T> ReadTransfer flow
    ConvertTransferConfig<TIn, TOut> ConvertTransfer operator
    ConditionTransferConfig<T> ConditionTransfer — (predicates are optional), onAcceptError?, onEmitError?
    CompositeTransferConfig<TIn, TOut> UniversalCompositeTransfer input, output
    PassBridgeConfig<T> PassBridge source, target, activated, linkStrategy?
    TransformBridgeConfig<TIn, TOut> TransformBridge source, target, operator, activated, linkStrategy?
    TransferBridgeConfig<TIn, TOut> TransferBridge source, target, middle, middleOwned, activated, linkStrategy?
    BridgeAggregatorConfig BridgeAggregator bridges, activated, owned
    BridgeSelectorConfig<TMap> BridgeSelector bridges, initialKey, activated, owned, syncWithChildren?
    BridgeMultiSelectorConfig<TMap> BridgeMultiSelector bridges, initialKeys, activated, owned, syncWithChildren?

    Async configs:

    Config For Required fields
    AsyncPollingProxyTransferConfig<T> Async polling transfers interval, activated
    AsyncPollingSourceTransferConfig<T> AsyncPollingSourceTransfer fetcher, interval, activated
    AsyncPollingFlowTransferConfig<T> AsyncPollingFlowTransfer flow, interval, activated
    AsyncIdlePollingTransferConfig<T> AsyncIdlePollingTransfer fetcher, timeout, interval, activated
    AsyncSinkTransferConfig<T> AsyncSinkTransfer callback, onError?, ordered?, maxConcurrency?, bufferSize?, onBufferOverflow?
    AsyncWriteTransferConfig<T> AsyncWriteTransfer flow, onError?, ordered?, maxConcurrency?, bufferSize?, onBufferOverflow?
    AsyncReadTransferConfig<T> AsyncReadTransfer flow
    AsyncConvertTransferConfig<TIn, TOut> AsyncConvertTransfer operator (AsyncOperatorInterface), onError?, maxConcurrency?, bufferSize?, onBufferOverflow?
    AsyncConditionTransferConfig<T> AsyncConditionTransfer — (predicates are optional, sync or async), onAcceptError?, onEmitError?, maxConcurrency?, bufferSize?, onBufferOverflow?
    AsyncStoredChannelTransferConfig<T> AsyncStoredChannelTransfer setup, destroy, onError?, onDestroyError?
    AsyncTransformBridgeConfig<TIn, TOut> AsyncTransformBridge source, target, operator, activated, onError?, linkStrategy?
    LinkConfig<TTargetTransfer> linkTransfers (async strategies) onError?

    All polling transfers support an optional tickerFactory?: TickerFactory to replace the default ticker (RAFTicker.factory).

    BackpressureConfig<T> (maxConcurrency?, bufferSize?, onBufferOverflow?) is shared by AsyncSinkTransfer, AsyncWriteTransfer, AsyncConvertTransfer, and AsyncConditionTransfer. See Backpressure.

    onLinkError in CompositeTransferBuilder.to() and finish() passes onError to linkTransfers for async-push rejection handling.

    Config For Required fields
    TickerConfig RAFTicker, IntervalTicker callback, interval?

    Many configs include optional error handlers (onError, onAcceptError, onEmitError, onDestroyError). Each handler receives (error: Error, source: TSource) where source is the transfer instance that triggered the error.

    npm i
    npm run test

    Transferum is licensed under the MIT License.

    Settings
    Member Visibility

    On This Page
    Transferum