Control Bus для .NET-маршрутов: останавливать, перезапускать и реагировать на события в рантайме

2026-08-21redb.Route

Работающая интеграция это не статичная штука. Деплою нужно слить очередь и остановить маршрут, прежде чем новая сборка перехватит управление. Нижестоящий сервис падает, и маршрут, который его долбит, должен отступить, а не громоздить ретраи. Оператору нужно приостановить одну ветку потока, не перезапуская весь процесс. Что-то ломается в три ночи, и дашборд дежурного должен загореться сам, а не потому что кто-то тейлил логи.

Apache Camel решил операционную половину этого давно, паттерном Control Bus: маршрутами управляют, отправляя сообщение на специальный эндпоинт, ровно так же, как двигают любое другое сообщение. redb.Route приносит этот паттерн в .NET, с набором команд Camel, и добавляет то, чего у Camel нет: консюмер, превращающий события жизненного цикла маршрута обратно в сообщения, которые можно роутить.

То есть направлений тут два. Наружу: сказать маршруту start, stop, suspend, resume или restart. Внутрь: подписаться на то, что маршруты делают, и реагировать. И то и другое обычные шаги пайплайна. Смотрим код.

Control Bus за минуту

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

// Остановить маршрут по id
.To("controlbus:route?routeId=orders&action=stop")

// То же, флюентно
.ControlBus(ControlBusAction.Stop, "orders")

Это шаг-продюсер. Когда до него доходит сообщение, названный маршрут останавливается. Всё остальное это вариации действия и один консюмер для событий.

Управляем маршрутом сообщением

Действие это глагол. Полный набор зеркалит Camel:

Действие	Эффект
`Start`	Запустить маршрут.
`Stop`	Остановить маршрут (консюмер снят, маршрут остаётся зарегистрированным).
`Suspend`	Приостановить маршрут.
`Resume`	Возобновить остановленный или приостановленный маршрут.
`Restart`	Stop, затем start через `restartDelay` (по умолчанию 1000 мс).
`Status`	Отдать статус маршрута.
`Stats`	Отдать статистику маршрута.
`Fail`	Остановить маршрут и пометить упавшим.

Во флюентной форме действие это enum, в URI это query-параметр action.

using redb.Route.ControlBus;

.ControlBus(ControlBusAction.Suspend, "payments")
.ControlBus(ControlBusAction.Restart, "orders", async: true)   // fire-and-forget

// URI-эквиваленты
.To("controlbus:route?routeId=payments&action=suspend")
.To("controlbus:route?routeId=orders&action=restart&async=true")

Раз это обычный шаг маршрута, управляющее действие ставится хвостом любого пайплайна. Таймер, приостанавливающий batch-маршрут вне рабочих часов. Вебхук, перезапускающий маршрут по смене конфига. Ветка choice, останавливающая консюмер при отравленном сообщении. Вы не дёргаете management-API снаружи, вы роутите сообщение, с теми же ретраями, обработкой ошибок и наблюдаемостью, что и остальной поток.

Маршрут может управлять собой

Передайте current как id маршрута, и действие нацелится на маршрут, отправляющий сообщение. Так маршрут реагирует на собственное состояние.

From("kafka://orders")
    .Process(CheckHealth)
    .Choice()
        .When(e => e.In.GetHeader<bool>("downstreamDown"))
            .ControlBus(ControlBusAction.Suspend, "current")   // отступить, перестать потреблять
    .End();

Маршрут, заметивший, что его downstream нездоров, приостанавливает сам себя вместо прокрутки отказов. Что-то внешнее, таймер или сообщение оператора, возобновит его позже.

Чего у Camel нет: реагировать на события маршрутов

Вот направление, которого обычно не хватает. У Camel Control Bus только на отправку: вы шлёте команды, но не подписываетесь на то, что происходит. redb.Route добавляет controlbus:notify, консюмер, эмитящий события жизненного цикла маршрутов и контекста как сообщения. Ставите его на сторону From и роутите события куда угодно.

From("controlbus:notify")
    .Process(e =>
    {
        var evt   = e.In.GetHeader<string>(ControlBusHeaders.Event);    // RouteStarted, RouteStopped, RouteErrored, ...
        var route = e.In.GetHeader<string>(ControlBusHeaders.RouteId);
        var when  = e.In.GetHeader<DateTimeOffset>(ControlBusHeaders.Timestamp);
    })
    .To("kafka://route-events");

События покрывают жизненный цикл маршрутов и самого контекста: RouteStarted, RouteStopped, RouteSuspending, RouteErrored, ContextStarting, ContextStarted, ContextStopping, ContextStopped, ExchangeTimedOut. Каждое сообщение несёт детали в заголовках: имя события, id затронутого маршрута, метку времени, а где уместно, ошибку, id обмена и затраченное время.

Фильтруйте до нужного параметрами events и routeId:

// Только отказы и остановки, только по маршруту orders
From("controlbus:notify?events=RouteErrored,RouteStopped&routeId=orders")
    .To("slack://alerts");

Интересное в том, что оба направления композятся. События на входе, команды на выходе, в одном маршруте: это самовосстанавливающийся цикл без внешнего control plane.

From("controlbus:notify?events=RouteErrored")
    .Process(e => e.In.SetHeader("failed", e.In.GetHeader<string>(ControlBusHeaders.RouteId)))
    .Delay(TimeSpan.FromSeconds(30))
    .ControlBus(ControlBusAction.Restart, "current");   // или пойманный id маршрута

Упавший маршрут эмитит событие, маршрут его потребляет, ждёт и перезапускает виновника. Логика супервизии это пайплайн, видимый и тестируемый как любой другой, а не захардкоженная политика, зарытая в движке.

Что это заменяет

Без Control Bus управление маршрутами в рантайме это груда одноразовой обвязки. Кастомный admin-контроллер, лезущий в движок остановить маршрут. BackgroundService, опрашивающий таблицу-флаг, чтобы понять, когда паузить. Логгер-аппендер, приклеенный к SDK алертинга, чтобы кто-то узнал об отказе. Каждое это отдельный механизм, со своим жизненным циклом, своими тестами, своим способом сломаться.

Control Bus превращает всё это в роутинг. Управляющие команды это сообщения To эндпоинт. События жизненного цикла это сообщения From эндпоинт. Они идут через тот же DSL, ту же обработку ошибок, те же OpenTelemetry-трейсы, что и бизнес-потоки, и приземляются там же, куда команда уже смотрит.

Деплой без простоя: приостановить и слить маршрут по сигналу, переключиться, возобновить.
Backpressure и circuit-breaking на уровне маршрута: маршрут приостанавливает себя, когда его downstream нездоров.
Операционные события как данные: слить RouteErrored и RouteStopped в Kafka, Slack, метрики, инцидент-трекер, аудит-лог.
Самовосстановление: notify на входе, restart на выходе, в одном маленьком маршруте.

Паритет с Camel и сверх него

Если пришли из Camel, командная сторона знакома: controlbus:route с routeId и action, плюс controlbus:language для управления выражением, только на отправку, стандартные глаголы. redb.Route повторяет этот набор команд. Консюмер controlbus:notify это добавка: Camel даёт EventNotifier как SPI, который вы реализуете на Java и регистрируете; redb.Route отдаёт ту же информацию как первоклассный эндпоинт, из которого вы роутите, без интерфейса для реализации, без проводки, просто From("controlbus:notify").

FAQ

Это отдельный пакет? Нет. Control Bus часть ядра redb.Route и зарегистрирован из коробки. Ставить нечего.

Остановка маршрута удаляет его? Нет. Stop и Suspend снимают консюмер, но оставляют маршрут зарегистрированным, так что Resume или Start возвращают его. Fail останавливает и помечает упавшим.

Может маршрут управлять другим маршрутом, не только собой? Да. Передайте целевой id вместо current.

Какие события доступны? Жизненный цикл маршрута (RouteStarted, RouteStopped, RouteSuspending, RouteErrored), жизненный цикл контекста (ContextStarting … ContextStopped) и ExchangeTimedOut. Фильтр через events= и routeId=.

Команда синхронная? По умолчанию да; передайте async=true (или async: true в DSL) для fire-and-forget, что заодно спасает маршрут от попытки синхронно остановить сам себя посреди обмена.

Где живёт

Control Bus поставляется внутри redb.Route на NuGet; DSL и набор событий controlbus:notify в документации. Это один из 30+ EIP-паттернов фреймворка, и как остальные, это шаг маршрута: тот же From → … → To, та же наблюдаемость. Разница в том, что сообщение, которое он несёт, это маршрут, рассказывающий, что он только что сделал, или вы, говорящий маршруту, что делать дальше.

Если было полезно, ⭐ на GitHub поможет другим это найти.

Другие мои статьи — redb.ru/articles, ещё — на Хабре.