Разработка 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 -> [ Transfer ] -> TИли вот так:
Transfer
// гетерогенный трансфер: TInput -> [ Transfer ] -> TOutputИменно в таком формате типов фабрики в библиотеке возвращают трансферы. Например:
const pushChannel: Transfer;
const converter: Transfer;
// содержат методы push(), subscribe()
const manualBuffer: Transfer;
// содержит методы push(), pull(), trigger()
// trigger() в данном случае нужен, чтобы сделать запушенное значение
// доступным для чтения через pull()
const pollingProxy: Transfer
// содержит методы pull(), subscribe(), trigger(), activate(), deactivate(), toggle()
// может быть связан с pullable-источником благодаря поддержке поллингаЭта «магия» работает в compile-time благодаря несколько замысловатой системе вычислимых типов:
type FeatureMap =
F extends { readonly isPushable: true } ? Pushable & { readonly isInput: true } :
F extends { readonly isPollingProxy: true } ? PollingProxy & { readonly isInput: true, readonly isOutput: true } :
F extends { readonly isPullable: true } ? Pullable & { readonly isOutput: true } :
F extends { readonly isSubscribable: true } ? Subscribable & { readonly isOutput: true } :
F extends { readonly isTriggerable: true } ? Triggerable :
F extends { readonly isGate: true } ? Gate :
F extends { readonly isAsyncPushable: true } ? AsyncPushable & { readonly isInput: true } :
F extends { readonly isAsyncPollingProxy: true } ? AsyncPollingProxy & { readonly isInput: true, readonly isOutput: true } :
F extends { readonly isAsyncPullable: true } ? AsyncPullable & { readonly isOutput: true } :
F extends { readonly isAsyncTriggerable: true } ? AsyncTriggerable :
unknown;
type UnionToIntersection = (U extends any ? (k: U) => void : never) extends ((k: infer I) => void) ? I : never;
type ResolveFeatures = UnionToIntersection;
type IsDuplexFeatures =
ResolveFeatures extends { readonly isInput: true, readonly isOutput: true }
? { readonly isDuplex: true }
: unknown;
type ExplicitTransfer = BaseTransferInterface
& ResolveFeatures
& IsDuplexFeatures;
export type Transfer<
TInOrAll,
TOutOrFeatures extends any[] | any,
TFeaturesOrUndefined extends any[] | undefined = undefined,
> = TFeaturesOrUndefined extends any[]
? ExplicitTransfer
: TOutOrFeatures extends any[]
? ExplicitTransfer
: 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();
const target = createSinkTransfer({ 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({
fetcher: () => Promise.resolve({ sensorId: 1, temperature: 25, humidity: 50 }),
interval: 50,
activated: true,
});
const sensor2 = createAsyncPollingSourceTransfer({
fetcher: () => Promise.resolve({ sensorId: 2, temperature: 26, humidity: 55 }),
interval: 50,
activated: true,
});
const sensorsAggregator = createMergeTransfer({
sources: [sensor1, sensor2],
});
const alertPipeline = CompositeTransferBuilder
.start(sensorsAggregator)
.to(createConditionTransfer({
shouldAccept: (d) => d.temperature > 95,
}))
.to(createThrottleTransfer({ interval: 5000 }))
.finish(createAsyncSinkTransfer({
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({ delay: 16 });
// Конвертер: KeyboardEvent -> код клавиши
const inputConverter = createConvertTransfer({
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. Если вы сталкивались с похожими задачами и решили их как-то иначе — буду рад прочитать о вашем опыте в комментариях.
