Из RxJS — и потокового мира вообще
Если вы пишете на RxJS, у нас для вас странная новость: вы уже наполовину думаете как мы. Операторы, конвейеры, «всё есть поток» — этот рефлекс здесь никуда не денется. Вторая половина новости приятнее: три вещи, с которыми вы годами боролись — глитчи combineLatest, зоопарк Subject-ов и утечки подписок, — в Continuum не требуют борьбы. Их невозможно написать.
Эта страница — длинная. Наливайте чай: мы разберём, почему Rx устроен так, как устроен, где именно расходятся наши дороги, и что произойдёт с каждым вашим привычным оператором.
Общая родословная: пол-идеи FRP
В 1997 году Конал Эллиотт и Пол Худак описали FRP: два типа — Behavior (значение во времени; в Continuum он называется State) и Stream (дискретные происшествия) — с точной математической семантикой композиции.
В 2009-м Эрик Мейер с командой в Microsoft выпустили Reactive Extensions. Rx взял у FRP слово «reactive» и алгебру операторов над потоками — и сознательно отбросил вторую половину: отдельный тип для «значений» и транзакционную одновременность. Получилась великолепная универсальная библиотека асинхронных потоков — но это другая вещь, со своими компромиссами.
Забавно, что потомки Rx в JS-мире по одному переизобретали отброшенное:
| Библиотека | Что вернула из FRP |
|---|---|
| Bacon.js (2012) | Property — поток с текущим значением: почти Behavior |
| Kefir (2014) | тоже Property, плюс диета по размеру |
| xstream (2016, для Cycle.js) | только «горячие» потоки — half-решение проблемы hot/cold |
| most.js | скорость и алгебраическая дисциплина |
| RxJS | BehaviorSubject + shareReplay(1) — ручная эмуляция Behavior |
Continuum не «переизобретает» — он идёт от исходной линии (Fran → Sodium), где обе половины идеи на месте. Отсюда все различия ниже.
Различие №1: у вас один тип, у нас два
В Rx всё — Observable. Даже то, что по природе не «поток происшествий», а «текущее значение»: имя пользователя, содержимое поля, флаг «загружается».
И вот тут начинается знакомая вам церемония:
// RxJS: "текущее значение" из потока — собери сам
const count$ = clicks$.pipe(
scan((n) => n + 1, 0),
startWith(0), // не забыть начальное!
shareReplay({ bufferSize: 1, refCount: true }), // не забыть replay!
);
// ...и всё равно молиться, что никто не подписался "слишком поздно"
// и что refCount не сбросит состояние, когда подписчики мигнут.Каждый кусочек этой цепочки — защита от того, что у потока нет текущего значения. startWith чинит «подписчик до первого события», shareReplay(1) чинит «подписчик после последнего», refCount чинит утечку от shareReplay, а баг «состояние сбросилось, потому что все отписались на мгновение» не чинит уже ничто — его просто ловят в проде.
В Continuum «текущее значение» — это отдельный тип, State (тот самый Behavior из FRP-литературы):
// Continuum
const count = clicks.accum(0, (_e, n) => n + 1);Всё. У count всегда есть значение. Нельзя подписаться «слишком поздно» — listen немедленно доставляет текущее, а sample() отвечает всегда, даже когда слушателей нет. Нельзя «сбросить состояние отпиской» — состояние принадлежит скоупу-владельцу, а не цепочке подписок. startWith, shareReplay, refCount, BehaviorSubject, дилемма hot/cold — этих слов в нашем словаре нет, потому что нет проблем, которые они решают.
Правило перевода простое:
Если ваш поток заканчивается на
scan/startWith/shareReplayили начинается сBehaviorSubject— это был не поток. Это был State, который заставили притворяться потоком.
Различие №2: глитчи — не баг вашего кода, а свойство модели Rx
Классика. Одно значение, две ветки, точка слияния:
// RxJS
const doubled$ = src$.pipe(map((x) => x * 2));
const squared$ = src$.pipe(map((x) => x * x));
combineLatest([doubled$, squared$]).subscribe(([d, s]) => {
console.log(d + s);
});
// src$ испускает 2: лог: 8 (4 + 4) ✓
// src$ испускает 3: лог: 10 !!! (6 + 4) ← ГЛИТЧ: новый doubled, старый squared
// лог: 15 (6 + 9) ✓combineLatest срабатывает на каждое обновление входа — и в момент, когда doubled$ уже увидел тройку, а squared$ ещё нет, наружу уезжает состояние мира, которого никогда не было. Если на этом суммировании висит запрос на сервер — вы только что отправили мусор.
Rx-ветераны знают заклинания: debounceTime(0), auditTime(0), переставить withLatestFrom, «просто не делать ромбов». Это дисциплина — то есть то, что однажды забудет джун в пятницу вечером.
В Continuum ромб — легальная, поощряемая топология:
// Continuum
const doubled = src.map((x) => x * 2);
const squared = src.map((x) => x * x);
const sum = combine(doubled, squared, (d, s) => d + s);
// src: 2 → 3. sum: 8 → 15. Одним шагом. Промежуточного "10" не существует.Каждое обновление — транзакция: атомарный момент, внутри которого узлы пересчитываются в порядке глубины графа (источники → производные → производные производных). Точка слияния физически не может сработать раньше, чем устаканятся оба входа. Это не оптимизация и не батчинг — это определение того, что значит «одновременно».
То же со слиянием событий: в Rx merge при одновременных входах отдаёт их «в каком-то порядке». У нас одновременность честная, и API требует сказать, что она означает:
// Continuum: два события в один момент — скажи, как их сложить
const delta = Stream.merge(plus, minus, (a, b) => a + b);Различие №3: подписки, за которыми не надо следить
Открыто любое Rx-приложение — увидите один из этих узоров:
// Узор 1: ручная бухгалтерия
private subs = new Subscription();
ngOnInit() { this.subs.add(a$.subscribe(...)); this.subs.add(b$.subscribe(...)); }
ngOnDestroy() { this.subs.unsubscribe(); }
// Узор 2: takeUntil-крест
private destroy$ = new Subject<void>();
a$.pipe(takeUntil(this.destroy$)).subscribe(...);
ngOnDestroy() { this.destroy$.next(); this.destroy$.complete(); }Angular придумал async-pipe, чтобы прятать эту бухгалтерию; сообщество — десяток линт-правил, чтобы её проверять. Потому что забытая отписка — баг №1 любой Rx-кодбазы: тихий, накапливающийся, всплывающий через месяц как «почему приложение жрёт память и дёргает обработчики умерших экранов».
В Continuum подписки регистрируются в дереве владения. Компонент построил привязку, регион, таймер — всё это записано за его поддеревом. Ушло поддерево (переключился роут, скрылся Show) — каскадно отписалось всё внутри. Не «нужно не забыть», а «нельзя забыть»:
function Ticker() {
const seconds = interval(1000).hold(0);
return <span>{seconds}</span>;
// размонтировали — таймер остановился, привязка отписалась. Всё.
}Единственное место, где отписка видима, — явный низкоуровневый listen, и там она одна строка: onCleanup(e.listen(handler)).
Словарик операторов
Рефлексы переносятся почти без перевода:
| RxJS | Continuum | Комментарий |
|---|---|---|
map, filter | e.map, e.filter | так же |
scan(f, init) | e.accum(init, f) | результат — сразу State |
startWith(x) + shareReplay(1) | e.hold(x) | одно слово вместо церемонии |
combineLatest | combine(a, b, f) | без глитчей |
withLatestFrom(b$) | s.at(e, f) | с точной одновременностью |
merge(a$, b$) | Stream.merge(a, b, f) | одновременные коалесцируются f |
race(a$, b$)-ish | a.or(b) | лево-приоритетный |
debounceTime(ms) | debounce(e, ms) | std |
throttleTime(ms) | throttle(e, ms) | std |
delay(ms) | delay(e, ms) | std |
interval(ms) | interval(ms) | std; Stream<number> |
distinctUntilChanged() | distinct(e) / dedupe(b) | std |
pairwise() | pairwise(e) | std |
take(1) / first() | e.once() | |
filter(() => flag) | e.when(flag) | флаг — State, не замыкание |
BehaviorSubject | state(init) | тип, а не костыль |
Subject | stream() | |
switchMap(fetch) | resource(e, fetch) | см. ниже |
subscribe | listen + onCleanup | или вообще привязка в JSX |
switchMap и асинхронность: наш ответ — данные
Половина реального RxJS-кода — это «поток запросов → switchMap → поток ответов» с отменой устаревшего:
// RxJS
query$.pipe(
debounceTime(300),
distinctUntilChanged(),
switchMap((q) =>
this.http.get(`/search?q=${q}`).pipe(
map((r) => ({ loading: false, data: r })),
startWith({ loading: true, data: null }),
catchError((e) => of({ loading: false, error: e })),
),
),
);Знакомо? Три состояния размазаны по startWith/catchError, отмена спрятана в семантике switchMap, и не дай бог перепутать его с mergeMap — получите гонку ответов, которую заметите не сразу.
Continuum сворачивает весь этот узор в одну функцию с честным типом:
// Continuum
const settled = debounce(query.updates, 300);
const results = resource(settled, (q) => api.search(q));
// State<Async<T>>: { status: "idle" | "loading" | "ok" | "error" }resource — это и есть «switchMap для UI»: последний запрос побеждает (ответы на устаревшие игнорируются по номеру, не по хрупкой отмене), loading/error — ветки значения, которые рендерятся обычным Dynamic. Ошибка — не исключение, летящее по пайпу и убивающее поток (помните сюрприз «после ошибки Observable мёртв»?), а данные.
Для нестандартных сценариев есть низкоуровневый perform:
const answers = perform(requests, (r) => api.send(r));
// Stream<Result<unknown, T>> — { ok: true, value } | { ok: false, error }Честное признание: у concatMap и exhaustMap (очередь запросов; игнорировать, пока летит текущий) готовых аналогов нет — это выражается через when/accum, но требует подумать. Если ваша задача — оркестровка очередей асинхронных операций, RxJS в этом сильнее.
Чего у нас нет — и почему это хорошо
- Hot vs cold. Наши Streams всегда «горячие» и мультикастные. Не бывает «каждый подписчик получает свой fetch», не нужен
share. - Complete/error как терминальные сигналы. Поток у нас не «умирает» — ни сам по себе после ошибки, ни по
complete. Ошибки — данные, конец жизни — уничтожение поддерева-владельца. - Backpressure. Мы UI-библиотека: быстрее, чем рендерится кадр, события не приходят. Если вам нужен настоящий backpressure — вам нужен не UI-фреймворк.
- Сто пятьдесят операторов.
bufferToggle,windowWhen,sequenceEqual… Наш std умещается в 2 kB. Экзотика пишется как обычная функция надStream— иногда это три строки, и они ваши. - Schedulers. Логическое время транзакций +
queueMicrotaskбраузера закрывают наши сценарии.
Где RxJS остаётся лучшим выбором
Симметричная честность. Берите Rx, а не нас, если у вас:
- не-UI пайплайны: обработка данных, сокеты с очередями, серверная оркестровка;
- сложная хореография времени: окна, буферы, zip нескольких потоков с выравниванием;
- очереди операций (
concatMap/exhaustMap-сценарии) как ядро задачи; - инфраструктура, уже стоящая на Rx (Angular, NestJS-микросервисы).
Continuum — не «RxJS получше». Это UI-фреймворк, в котором событийная алгебра — одна из двух половин модели, а не вся модель.
Итог в одну строку
Как RxJS — но с настоящими значениями, без глитчей и с автоматической отпиской. И сеть подключена к DOM напрямую.
Куда дальше:
- Что такое FRP — на пальцах — если хочется понять модель с нуля;
- Streams и States — строгие определения обеих половин;
- Транзакции и время — как именно устроено «без глитчей»;
- История и контекст — откуда растут и Rx, и мы.
Незнакомый термин? Вся терминология этой документации объяснена в глоссарии.