Smoren
Разработка

Разработка TypeScript-библиотеки для построения реактивных графов распространения и обработки данных

Предыстория

Все началось с рабочей задачи по реализации весьма специфичной web-панели мониторинга и управления различным оборудованием, в которой нужно было получать данные и отправлять команды по разнообразным сценариям (периодический опрос, подписка на websocket-события, https-запросы и т.д.), а также выводить состояние и графики в реальном времени, динамически комбинируя различные источники данных в разных виджетах.

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

В первой итерации с использованием RxJS получилось много неструктурированного и сложно поддерживаемого кода, так как архитектура на RxJS вынуждала императивно описывать перестроение топологии при изменении правил маршрутизации «на лету». В нашем специфичном кейсе с динамическими виджетами это приводило к сайд-эффектам и сложностям в отладке. Кроме того, местами размывалась строгая типизация и мы лишались compile-time гарантий. Так появилась идея разработать собственное решение на основе альтернативной концепции — модели графа потоков данных.

В результате получившаяся модель продемонстрировала предсказуемое поведение и низкую связность компонентов, а кодовая база сократилась на ~30% и приобрела более декларативный вид. Убедившись в эффективности решения, мы решили оформить его в виде отдельной библиотеки с открытым исходным кодом — Transferum.

Разработка TypeScript-библиотеки для построения реактивных графов распространения и обработки данных

И что, получилась просто еще одна реактивная библиотека?

Не совсем. Классические FRP-библиотеки (RxJS, Bacon, Most) построены вокруг единственного примитива (Observable). Transferum же основан на композиции различных типов узлов с явно определенным поведением.

Каждый узел в графе потоков распространения данных явно декларирует свои способности: может ли он принимать данные через push, отдавать через pull, распространять полученный сигнал подписчикам, опрашивать источник, фильтровать, блокировать поток и т.д.

Объявленные узлом возможности являются одновременно флагами для использования в runtime и compile-time гарантиями наличия соответствующих методов, определяющих его поведение.

Ключевая идея: поведение системы описывается как композиция независимых возможностей, которые одновременно определяют тип, реализацию и правила взаимодействия.

Transferum предоставляет четыре слоя абстракции:

  1. Трансферы — узлы графа (каналы, поллеры, мапперы, буферы, разветвители и концентраторы, реализации debounce, throttle, switchMap и т.д.).
  2. Мосты — ребра графа — управляемые вентили между узлами с динамической маршрутизацией и гейтингом.
  3. Операторы — stateless-трансформаторы и фильтры данных (используются трансферами, отвечающими за конвертацию данных).
  4. Билдеры — fluent-конструкторы композитных трансферов из цепочек трансферов-примитивов.

Концептуальная и архитектурная основа — capability flags system. Каждый трансфер реализует CommunicationContractInterface — набор булевых флагов, определяющих его возможности.

Флаги isPushable, isPullable, isSubscribable, isGate и другие — это не просто свойства объекта. Это метаданные, которые:

  • Определяют TypeScript-интерфейс трансфера на этапе компиляции.
  • Управляют стратегией связывания с другими трансферами в рантайме (с помощью функции linkTransfers()).
  • Обеспечивают совместимость в билдерах без приведений типов.

Один набор флагов — три потребителя. Это единый источник истины для всей системы.

Когда флаг равен true, соответствующий метод входит в TypeScript-интерфейс трансфера. Это позволяет предоставлять трансфер пользователю вот так:

Transfer<T, [Pushable, Pullable, Subscribable, Triggerable]> // гомогенный трансфер: T -> [ Transfer ] -> T

Или вот так:

Transfer<TInput, TOutput, [Pushable, Pullable, Subscribable, Triggerable]> // гетерогенный трансфер: TInput -> [ Transfer ] -> TOutput

Именно в таком формате типов фабрики в библиотеке возвращают трансферы. Например:

const pushChannel: Transfer<number, [Pushable, Subscribable]>; const converter: Transfer<number, string, [Pushable, Subscribable]>; // содержат методы push(), subscribe() const manualBuffer: Transfer<number, [Pushable, Pullable, Triggerable]>; // содержит методы push(), pull(), trigger() // trigger() в данном случае нужен, чтобы сделать запушенное значение // доступным для чтения через pull() const pollingProxy: Transfer<T, [PollingProxy, Pullable, Subscribable, Triggerable, Gate]> // содержит методы pull(), subscribe(), trigger(), activate(), deactivate(), toggle() // может быть связан с pullable-источником благодаря поддержке поллинга

Эта «магия» работает в compile-time благодаря несколько замысловатой системе вычислимых типов:

type FeatureMap<TIn, TOut, F> = F extends { readonly isPushable: true } ? Pushable<TIn> & { readonly isInput: true } : F extends { readonly isPollingProxy: true } ? PollingProxy<TIn> & { readonly isInput: true, readonly isOutput: true } : F extends { readonly isPullable: true } ? Pullable<TOut> & { readonly isOutput: true } : F extends { readonly isSubscribable: true } ? Subscribable<TOut> & { readonly isOutput: true } : F extends { readonly isTriggerable: true } ? Triggerable : F extends { readonly isGate: true } ? Gate : F extends { readonly isAsyncPushable: true } ? AsyncPushable<TIn> & { readonly isInput: true } : F extends { readonly isAsyncPollingProxy: true } ? AsyncPollingProxy<TIn> & { readonly isInput: true, readonly isOutput: true } : F extends { readonly isAsyncPullable: true } ? AsyncPullable<TOut> & { readonly isOutput: true } : F extends { readonly isAsyncTriggerable: true } ? AsyncTriggerable : unknown; type UnionToIntersection<U> = (U extends any ? (k: U) => void : never) extends ((k: infer I) => void) ? I : never; type ResolveFeatures<TIn, TOut, Features extends any[]> = UnionToIntersection<{ [K in keyof Features]: FeatureMap<TIn, TOut, Features[K]> }[number]>; type IsDuplexFeatures<Features extends any[]> = ResolveFeatures<any, any, Features> extends { readonly isInput: true, readonly isOutput: true } ? { readonly isDuplex: true } : unknown; type ExplicitTransfer<TIn, TOut, Features extends any[]> = BaseTransferInterface & ResolveFeatures<TIn, TOut, Features> & IsDuplexFeatures<Features>; export type Transfer< TInOrAll, TOutOrFeatures extends any[] | any, TFeaturesOrUndefined extends any[] | undefined = undefined, > = TFeaturesOrUndefined extends any[] ? ExplicitTransfer<TInOrAll, TOutOrFeatures, TFeaturesOrUndefined> : TOutOrFeatures extends any[] ? ExplicitTransfer<TInOrAll, TInOrAll, TOutOrFeatures> : never;

Архитектурные инварианты

1. Трансферы не знают своих соседей

Трансфер определяет своё поведение (push, pull, subscribe и др.), но никогда не ссылается и не проверяет класс другого трансфера. Он не знает, что является upstream или downstream — лишь выполняет свой контракт. Пользователь может создать свой трансфер, объявить и реализовать его возможности — и он органично и бесшовно впишется в экосистему.

2. Мосты не знают конкретных реализаций

Мост инспектирует capability flags, а не имена классов. Нет цепочки instanceof, нет переключения по имени класса. Любой output-трансфер может быть соединен с любым input-трансфером — при условии совместимости их флагов, о чем мы поговорим чуть ниже. Это применимо и к тем узлам, которые еще не существуют и будут созданы пользователем.

3. Значение undefined никогда не распространяется

В Transferum undefined означает «нет данных», а не «пустое значение». Оно подавляется на уровне внутренней реализации менеджера подписок — подписчики никогда не уведомляются с undefined. При этом для явных маркеров пустых значений можно использовать null. Мы сознательно пошли на этот компромисс, чтобы избежать runtime-оверхеда и сохранить нативную скорость работы на плотных потоках данных.

Связывание трансферов

Функция linkTransfers(lhs, rhs) соединяет output-трансфер (lhs) с input-трансфером (rhs) с автоматическим выбором стратегии связывания на основе возможностей этих трансферов:

  • isSubscribable → isPushable (реактивная подписка);
  • isPullable → isPollingProxy (активный опрос);
  • isSubscribable → isAsyncPushable (реактивная подписка + асинхронный push);
  • isAsyncPullable → isAsyncPollingProxy (активный асинхронный опрос асинхронного pull-источника);
  • isPullable → isAsyncPollingProxy (активный асинхронный опрос синхронного pull-источника).

Protocol-oriented design: механизм не спрашивает «какой это класс?» — он выясняет, какие у него есть возможности. Любая пара трансферов с совместимыми возможностями является linkable. Добавление нового класса трансфера требует только объявления его флагов и реализации соответствующих методов — как связать его с другим трансфером, связующий алгоритм разберется сам.

import { createPushChannelTransfer, createSinkTransfer, linkTransfers } from 'transferum'; const source = createPushChannelTransfer<number>(); const target = createSinkTransfer<number>({ callback: (x) => console.log(x) }); const link = linkTransfers(source, target); source.push(22); link.unsubscribe(); // разорвать связь

Sync и async в одной экосистеме

Синхронные и асинхронные трансферы сосуществуют и могут быть связаны между собой. linkTransfers() предпочитает sync-связывание, когда это возможно, а async-стратегии применяет только когда sync неприменим. Нет отдельного «асинхронного мира».

Поддержка backpressure

Ряд асинхронных трансферов (AsyncSinkTransfer, AsyncWriteTransfer, AsyncConvertTransfer, AsyncConditionTransfer) поддерживают необязательные поля в конфигурации: maxConcurrency, bufferSize и onBufferOverflow — для ограничения параллельных async-операций, очереди избыточных данных и graceful-обработки переполнения. По умолчанию — неограниченная обработка, без буферизации.

Локальная обработка ошибок

Transferum использует единую модель обработки ошибок для всех трансферов. Каждый трансфер, который может столкнуться с runtime-ошибкой, принимает опциональный onError-хэндлер в своей конфигурации.

А теперь — к примерам использования

Вот так можно просто и декларативно описать опрос и агрегирование данных из нескольких источников:

import { createMergeTransfer, createConditionTransfer, createThrottleTransfer, createAsyncSinkTransfer, createAsyncPollingSourceTransfer, CompositeTransferBuilder, } from 'transferum'; const sensor1 = createAsyncPollingSourceTransfer<SensorData>({ fetcher: () => Promise.resolve({ sensorId: 1, temperature: 25, humidity: 50 }), interval: 50, activated: true, }); const sensor2 = createAsyncPollingSourceTransfer<SensorData>({ fetcher: () => Promise.resolve({ sensorId: 2, temperature: 26, humidity: 55 }), interval: 50, activated: true, }); const sensorsAggregator = createMergeTransfer<SensorData>({ sources: [sensor1, sensor2], }); const alertPipeline = CompositeTransferBuilder .start(sensorsAggregator) .to(createConditionTransfer<SensorData>({ shouldAccept: (d) => d.temperature > 95, })) .to(createThrottleTransfer({ interval: 5000 })) .finish(createAsyncSinkTransfer<SensorData>({ callback: (sensorData) => { console.warn(`Alert! Sensor ${sensorData.sensorId}: temperature is ${sensorData.temperature}`); } })); sensorsAggregator.subscribe((sensorData) => console.log('Debugging', sensorData));

А вот как можно организовать динамический роутинг:

import { createPushStoredChannelTransfer, createBridgeMultiSelector, createBridgeSelector, createPassBridge, } from 'transferum'; // Вариант 1. Выбор одного реципиента 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: 'thermostat', // выбор по умолчанию activated: true, owned: true, }); commandRouter.select('lock'); // выберем другой вариант // Вариант 2. Выбор нескольких реципиентов const loggersRouter = 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, }); loggersRouter.check('elk'); // теперь выбраны все три loggersRouter.uncheck('prometheus'); // выбраны elk, sentry loggersRouter.select(['prometheus', 'sentry']); // выбраны prometheus, sentry

А вот пример c организацией игровой механики:

import { createDebounceTransfer, createConvertTransfer, createMapOperator, createBridgeSelector, createPassBridge, linkTransfers, } from 'transferum'; // Источник: debounce нажатий клавиш (16ms ~ 1 кадр при 60 FPS) const rawInputSource = createDebounceTransfer<KeyboardEvent>({ delay: 16 }); // Конвертер: KeyboardEvent -> код клавиши const inputConverter = createConvertTransfer<KeyboardEvent, string>({ operator: createMapOperator((event) => event.code), }); // Целевые системы-потребители пользовательских событий 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}`) }; // Роутер: динамическое переключение между игровыми контекстами const gameplayRouter = createBridgeSelector({ bridges: { driving: createPassBridge({ source: inputConverter, target: carSystem }), flying: createPassBridge({ source: inputConverter, target: planeSystem }), ui: createPassBridge({ source: inputConverter, target: menuSystem }), }, initialKey: 'driving', // игрок начинает в машине activated: true, }); // Связывание источника с конвертером с автовыбором стратегии linkTransfers(rawInputSource, inputConverter); // Демонстрация: // Игрок в машине нажимает "W" rawInputSource.push(new KeyboardEvent('keydown', { code: 'KeyW' })); // Через 16ms: 🚗 Car: KeyW // ================== // Переключаемся на самолёт gameplayRouter.select('flying'); // Та же клавиша теперь управляет самолётом rawInputSource.push(new KeyboardEvent('keydown', { code: 'KeyW' })); // Через 16ms: ✈️ Plane: KeyW

Когда имеет смысл попробовать Transferum

Библиотека подойдет для:

  • TypeScript-first проектов — благодаря максимально строгой типизации и compile-time вычислению доступных методов любого трансфера на основе объявленных у него флагов возможностей.
  • Работы с pull-based источниками данных — polling API, датчиков, хранилищ с PollingProxy.
  • Смешанных sync/async пайплайнов — в единой модели без ручного преобразования.
  • Явного flow control — gates, bridges, selectors для runtime-маршрутизации.
  • Game development / IoT — frame-aligned tickers, idle polling, sensor aggregation.
  • Устойчивой обработки ошибок — локальная, non-fatal обработка: одна стадия не убивает пайплайн при условии переданного в конфиге обработчика ошибок, ничего не подавляется молча.

Результаты и планы

  • Библиотека уже используется в двух наших внутренних проектах и показывает свою эффективность. Код доступен на GitHub под лицензией MIT, библиотека не имеет внешних зависимостей и поставляется с подробной документацией (README + API Reference).
  • В одном из проектов граф состоит из ~80 узлов и стабильно обрабатывает несколько сотен событий в секунду без деградации. На основе этих данных в том числе рендерится 3D-сцена в Babylon.js со стабильным фреймрейтом ~60 FPS без микрофризов.
  • Тесты библиотеки покрывают не только отдельные трансферы, но и поведение системы в динамике: переподключение мостов, обработку ошибок в длинных асинхронных цепочках, а также разнообразные граничные случаи. Покрытие — 100%.
  • В дальнейшем планируем реализовать хуки и утилиты для более удобного и нативного использования Transferum с Vue и React. Если они окажутся в достаточной мере переиспользуемыми, оформим в отдельные пакеты-адаптеры.

Буду рад, если вы заглянете в репозиторий, попробуете библиотеку в деле и поделитесь замечаниями — обратная связь поможет сделать Transferum лучше.

P. S. Если вы сталкивались с похожими задачами и решили их как-то иначе — буду рад прочитать о вашем опыте в комментариях.