Building explicit data-flow graphs in TypeScript: Introducing Transferum

Hi, dev.to community! Today I want to share my open-source project — Transferum. It is a TypeScript library designed for building graph-based interaction models between components using reactive data streams and flexible routing.

Why yet another reactive library?

Unlike traditional reactive frameworks (RxJS, Most.js, etc.) that revolve around a single abstract primitive (the Observable stream), Transferum is built on the concept of capabilities.

Every node in the graph — called a Transfer — explicitly declares what it can do: push data, pull data, filter, buffer, poll, or block the stream. These capabilities serve a dual purpose:

• Compile-time guarantees: TypeScript knows exactly which methods are available at each pipeline stage. No type casting or unsafe runtime checks needed.
• Runtime values: They dictate how nodes connect dynamically.

Key Idea: System behavior is described as a composition of independent capabilities that simultaneously define the type, implementation, and interaction rules.

The Four Layers of Abstraction

Transferum structures data flow using four clean abstractions:

1. Transfers: Graph nodes (channels, buffers, debounce/throttle, polling, merge/split, gate/routing, adapters).
2. Bridges: Controlled connections between nodes supporting dynamic routing and gating.
3. Operators: Stateless transformers and data filters (both synchronous and asynchronous).
4. Builders: Chain constructors (including async-powered variations).

Architecture: Capability Flags Under the Hood

The architectural foundation relies on a system of capability flags like isPushable, isPullable, isSubscribable, and isGate. They are not just plain object properties. They act as computed metadata that:

• Shape the TypeScript interface of a Transfer during compilation.
• Manage the binding strategy inside the linkTransfers() engine at runtime.
• Ensure strict compatibility within builders without type assertion.

Dynamic Binding Strategies

The linkTransfers(lhs, rhs) function connects an output transfer (LHS) to an input transfer (RHS) by automatically resolving the strategy based on flags:

• isSubscribable → isPushable: Reactive subscription.
• isPullable → isPollingProxy: Active polling.
• isSubscribable → isAsyncPushable: Subscription + asyncPush.
• isAsyncPullable → isAsyncPollingProxy: Async polling.
• isPullable → isAsyncPollingProxy: Sync-pull → async fetcher.

This is protocol-oriented design. The engine never asks "What class is this?" — it asks "What capabilities does it have?".

Core Architectural Invariants

To keep the codebase predictable, Transferum enforces three strict invariants:

1. Transfers don't know their neighbors

A transfer defines its own behavior (push, pull, etc.) but never references or checks the class of another transfer. It remains completely agnostic of what is upstream or downstream.

2. Bridges don't know concrete implementations

Bridges inspect capability flags, not class names. No instanceof chains, no RTTI, no string-based switch cases. Any transfer with the correct flags is instantly bridgeable.

3. The undefined value never propagates

In Transferum, undefined explicitly means "no data", not "empty value". It is suppressed inside the subscription manager, meaning subscribers are never triggered with undefined. You can use null for explicit empty markers.

Mixing Sync and Async with Flow Control

Synchronous and asynchronous transfers coexist natively. linkTransfers() prioritizes synchronous binding whenever possible and falls back to asynchronous strategies only when necessary. There is no isolated "async world".

Backpressure for Async Transfers

Four built-in async transfers (AsyncSinkTransfer, AsyncWriteTransfer, AsyncConvertTransfer, AsyncConditionTransfer) support optional configuration for:

• maxConcurrency — limiting concurrent async operations.
• bufferSize — queueing redundant incoming data.
• onBufferOverflow — graceful overflow handling strategies.

By default, it is backward-compatible (unbounded processing, no buffering).

Localized Error Handling

Failures shouldn't crash your entire data tree. Transferum implements a unified error handling model. Every transfer prone to runtime errors accepts an optional onError handler in its configuration.

It ensures fail-safe execution: a single failing stage doesn't kill the whole pipeline, broken pollers halt gracefully, and no error is silently swallowed.

Practical Examples

1. Data Aggregation (IoT Sensor Style)

Here is how you can merge multi-source polling feeds into a single async pipeline:

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

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

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

const aggregator = createMergeTransfer<SensorData>({
  sources: [tempSensor, humiditySensor], // Strongly typed merging
});

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

2. Dynamic Traffic Routing

You can toggle bridges on the fly to route metrics or commands:

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

// Scenario A: Route to a single target
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'); // Instantly switches stream to thermostat

// Scenario B: Broadcast to multiple targets
const metricsChannel = createPushStoredChannelTransfer<Metric>();

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

metricsChannel.push({ name: 'request_latency', value: 150 }); // Dispatched to Prometheus and ELK

When should you use Transferum?

Transferum shines in:

• TypeScript-first setups: Where you need absolute type safety based on compile-time capabilities.
• Mixed Sync/Async Pipelines: Eliminating manual promise wrapping or stream transformations.
• Pull-based Data Fetching: Seamless integration with polling APIs, hardware sensors, or state storages via PollingProxy.
• Explicit Flow Control: Building complex routing tables, conditional gates, and dynamic runtime bridges.
• Game Dev & IoT: Handling frame-aligned tickers, idle polling, and complex event aggregation.

Status & Open Source

• Zero Dependencies
• 100% Test Coverage (over 2000 test cases)
• MIT License
• Comprehensive documentation included

Get started via npm:

npm i transferum

The project is actively developed on GitHub: Smoren/transferum-ts.

I would highly appreciate your feedback, GitHub stars ⭐, and discussions! What patterns do you use for explicit graph routing in your projects? Let's talk in the comments!