Реактивные графы данных в TypeScript: архитектура библиотеки Transferum

Реактивные графы данных в TypeScript: архитектура библиотеки Transferum

Проектирование сложных пользовательских интерфейсов и систем реального времени на языке TypeScript часто требует обработки множества разнородных потоков данных. В приложениях мониторинга оборудования, панелях управления IoT-устройствами, торговых терминалах и игровых движках данные поступают по разным каналам: через периодический опрос HTTP-эндпоинтов, постоянные WebSocket-соединения или события пользовательского ввода. При этом в зависимости от выбранного режима работы приложения маршруты распространения этих данных должны динамически перестраиваться.

Традиционное применение библиотек функционального реактивного программирования (FRP), таких как RxJS, в подобных сценариях приводит к усложнению кодовой базы. Разработчикам приходится вручную контролировать жизненный цикл подписок, предотвращать утечки памяти и выстраивать операторы переключения потоков. Открытая TypeScript-библиотека Transferum предлагает альтернативную архитектурную модель — построение явных реактивных графов распространения данных.

Ограничения классического FRP при работе с динамической топологией

Классические FRP-библиотеки построены вокруг единого базового примитива — потока событий (Observable). Парадигма подразумевает, что данные непрерывно текут от источника к подписчикам через цепочки операторов трансформации.

Однако когда архитектура приложения требует динамического изменения самой топологии связей в реальном времени — например, переключения источника метрик для виджета или временного перекрытия потока — использование Observable заставляет писать код с побочными эффектами. Императивное создание и разрыв подписок в сложных графах приводит к рассинхронизации состояния, сложной отладке и постепенной утрате строгой типизации TypeScript на стыках динамических каналов.

Transferum отказывается от единственного примитива в пользу модели графа, где узлы (трансферы) обладают явным независимым поведением, а связи между ними (мосты) управляются динамически.

Четыре слоя абстракции: трансферы, мосты, операторы и билдеры

Архитектура библиотеки Transferum разделена на четыре чётких слоя ответственности:

1. Трансферы (Transfers): независимые узлы графа. Это могут быть каналы связи, поллеры, буферы, разветвители или мапперы. Каждый трансфер выполняет свою локальную задачу и не содержит ссылок на соседние узлы.
2. Мосты (Bridges): управляемые ребра графа, соединяющие трансферы. Мосты выступают в роли вентилей: они могут активироваться, деактивироваться и переключать маршрутизацию данных без пересоздания самих узлов.
3. Операторы (Operators): чистые функции без состояния (stateless), предназначенные для фильтрации и преобразования значения внутри трансферов.
4. Билдеры (Builders): удобные конструктивы с текучим интерфейсом (fluent interface) для сборки сложных цепочек трансферов в единый конвейер.

Система флагов возможностей capability flags как единственный источник истины

Ключевой технической идеей Transferum является система флагов возможностей (capability flags system). Каждый узел графа объявляет набор булевых свойств, описывающих его поведение:

• isPushable — способность принимать данные в синхронном режиме;
• isPullable — способность отдавать текущее значение по запросу;
• isSubscribable — возможность оформления реактивной подписки;
• isPollingProxy — способность выполнять периодический опрос источника;
• isGate — наличие функций управления прохождением потока.

Флаги возможностей выполняют роль единого источника истины для всей системы. На этапе компиляции TypeScript они участвуют в вычислении публичного типа трансфера, открывая только имеющиеся методы (например, push(), pull() или subscribe()). В процессе выполнения (runtime) эти же флаги используются алгоритмом автоматического связывания узлов.

Архитектурные инварианты: изоляция узлов, мосты и обработка undefined

Для обеспечения надежности и предсказуемости поведения графа в Transferum заложены три строгих инварианта:

1. Полная изоляция узлов: трансферы никогда не проверяют класс или тип соседних узлов. Узел знает только свои собственные возможности и выполняет свой локальный контракт.
2. Инспекция флагов вместо классов: мосты и алгоритмы связывания проверяют только булевые флаги возможностей, а не имена классов. Это позволяет разработчикам создавать собственные типы трансферов, которые бесшовно интегрируются в существующую систему.
3. Подавление значения undefined: в концепции Transferum значение undefined интерпретируется как полное отсутствие данных, а не как допустимое значение. Менеджер подписок автоматически подавляет undefined, не уведомляя подписчиков. Для передачи пустых состояний применяется значение null.

Автоматическое связывание узлов и сопряжение синхронных и асинхронных потоков

Соединение двух трансферов выполняется с помощью функции linkTransfers(lhs, rhs). Функция анализирует флаги левого (выходного) и правого (входного) узлов и автоматически выбирает оптимальную стратегию связывания:

• При наличии isSubscribable и isPushable формируется прямая реактивная подписка.
• При наличии isPullable у источника и isPollingProxy у приемника запускается активный опрос.
• При сочетании синхронных и асинхронных флагов выстраивается асинхронный канал с обработкой Promise.

Пример декларативного объединения данных от нескольких сенсоров:

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

// Создаем два поллера для опроса сенсоров
const sensor1 = createAsyncPollingSourceTransfer({
  fetcher: () => fetch('/api/sensor1').then(r => r.json()),
  interval: 100,
  activated: true
});

const sensor2 = createAsyncPollingSourceTransfer({
  fetcher: () => fetch('/api/sensor2').then(r => r.json()),
  interval: 100,
  activated: true
});

// Объединяем потоки в единый узел
const mergedSensors = createMergeTransfer({
  sources: [sensor1, sensor2]
});

// Строим цепочку фильтрации и реагирования
OutputPipelineBuilder
  .start(mergedSensors)
  .to(createConditionTransfer({
    shouldAccept: (data) => data.temperature > 80
  }))
  .finish(createAsyncSinkTransfer((data) => {
    console.warn("Alert! Sensor:", data.sensorId, "temp:", data.temperature);
  }));

Управление давлением backpressure, обработка ошибок и сферы применения

При работе с высоконагруженными асинхронными потоками асинхронные трансферы Transferum поддерживают параметры управления давлением (backpressure): maxConcurrency для ограничения параллельных вызовов, bufferSize для регулирования очереди и onBufferOverflow для стратегии обработки переполнения.

Каждый узел принимает локальный обработчик ошибок onError. Возникновение исключения на одной стадии не разрушает весь граф и не приводит к молчаливому подавлению ошибок.

Библиотека показала свою эффективность в задачах визуализации реального времени, где граф из 80 узлов обрабатывает сотни событий в секунду, обеспечивая рендеринг 3D-сцены в Babylon.js со стабильной частотой 60 FPS. Исходный код доступен под лицензией MIT в реестре пакетов JSR.