Разработка TypeScript-библиотеки для построения реактивных графов распространения и обработки данных
Предыстория
Все началось с рабочей задачи по реализации весьма специфичной web-панели мониторинга и управления различным оборудованием, в которой нужно было получать данные и отправлять команды по разнообразным сценариям (периодический опрос, подписка на websocket-события, https-запросы и т.д.), а также выводить состояние и графики в реальном времени, динамически комбинируя различные источники данных в разных виджетах.
Кроме того, у панели было предусмотрено несколько настраиваемых режимов работы, которые переключались по запросу пользователя, что требовало массового управления маршрутами потоков данных внутри системы.
В первой итерации с использованием RxJS получилось много неструктурированного и сложно поддерживаемого кода, так как архитектура на RxJS вынуждала императивно описывать перестроение топологии при изменении правил маршрутизации «на лету». В нашем специфичном кейсе с динамическими виджетами это приводило к сайд-эффектам и сложностям в отладке. Кроме того, местами размывалась строгая типизация и мы лишались compile-time гарантий. Так появилась идея разработать собственное решение на основе альтернативной концепции — модели графа потоков данных.
В результате получившаяся модель продемонстрировала предсказуемое поведение и низкую связность компонентов, а кодовая база сократилась на ~30% и приобрела более декларативный вид. Убедившись в эффективности решения, мы решили оформить его в виде отдельной библиотеки с открытым исходным кодом — Transferum.
И что, получилась просто еще одна реактивная библиотека?
Не совсем. Классические FRP-библиотеки (RxJS, Bacon, Most) построены вокруг единственного примитива (Observable). Transferum же основан на композиции различных типов узлов с явно определенным поведением.
Каждый узел в графе потоков распространения данных явно декларирует свои способности: может ли он принимать данные через push, отдавать через pull, распространять полученный сигнал подписчикам, опрашивать источник, фильтровать, блокировать поток и т.д.
Объявленные узлом возможности являются одновременно флагами для использования в runtime и compile-time гарантиями наличия соответствующих методов, определяющих его поведение.
Ключевая идея: поведение системы описывается как композиция независимых возможностей, которые одновременно определяют тип, реализацию и правила взаимодействия.
Transferum предоставляет четыре слоя абстракции:
Трансферы — узлы графа (каналы, поллеры, мапперы, буферы, разветвители и концентраторы, реализации debounce, throttle, switchMap и т.д.).
Мосты — ребра графа — управляемые вентили между узлами с динамической маршрутизацией и гейтингом.
Операторы — stateless-трансформаторы и фильтры данных (используются трансферами, отвечающими за конвертацию данных).
Билдеры — 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
Именно в таком формате типов фабрики в библиотеке возвращают трансферы. Эта «магия» работает в compile-time благодаря несколько замысловатой системе вычислимых типов.
Архитектурные инварианты
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. Добавление нового класса трансфера требует только объявления его флагов и реализации соответствующих методов — как связать его с другим трансфером, связующий алгоритм разберется сам.
Sync и async в одной экосистеме
Синхронные и асинхронные трансферы сосуществуют и могут быть связаны между собой. linkTransfers() предпочитает sync-связывание, когда это возможно, а async-стратегии применяет только когда sync неприменим. Нет отдельного «асинхронного мира».
Поддержка backpressure
Ряд асинхронных трансферов (AsyncSinkTransfer, AsyncWriteTransfer, AsyncConvertTransfer, AsyncConditionTransfer) поддерживают необязательные поля в конфигурации: maxConcurrency, bufferSize и onBufferOverflow — для ограничения параллельных async-операций, очереди избыточных данных и graceful-обработки переполнения. По умолчанию — неограниченная обработка, без буферизации.
Локальная обработка ошибок
Transferum использует единую модель обработки ошибок для всех трансферов. Каждый трансфер, который может столкнуться с runtime-ошибкой, принимает опциональный onError-хэндлер в своей конфигурации.
А теперь — к примерам использования
Вот так можно просто и декларативно описать опрос и агрегирование данных из нескольких источников.
А вот как можно можно организовать роутинг в игровой механике.
А вот пример сложного цепочечного трансфера, реализующего логику автоматизации обработки событий и принятия решения в торговле на бирже.
Когда имеет смысл попробовать 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. Если вы сталкивались с похожими задачами и решили их как-то иначе — буду рад прочитать о вашем опыте в комментариях.
