Устройство

Где стоит ruststream-sqlx: между рантаймом RustStream и пулом sqlx сервиса. SQL строится один раз на подписку, а outbox работает рядом с любым брокером.

Схема текстом

1. Ваш сервис: #[subscriber] обработчики-подписчики; RustStream::new(..) .with_broker(..)
2. Ядро RustStream: Рантайм: диспетчер, воркеры, завершение, публикация
3. ruststream-sqlx: InboxSubscriber цикл захвата; InboxDelivery завершает строку; Repository, Routed вставляют строки; InboxQueue<Row> дескриптор; SqlxBroker конфигурация; Dialect SQL каждой базы; Outbox слои, любой брокер
4. sqlx: Pool<DB> строится при подключении
5. #[subscriber] обработчики-подписчики → Рантайм: диспетчер, воркеры, завершение, публикация: монтируются
6. Рантайм: диспетчер, воркеры, завершение, публикация → InboxSubscriber цикл захвата: опрашивает
7. InboxSubscriber цикл захвата → Pool<DB> строится при подключении
8. Repository, Routed вставляют строки → Pool<DB> строится при подключении
9. Pool<DB> строится при подключении → PostgreSQL, MySQL, MariaDB или SQLite: драйвер
10. С фичей testing TestApp подключается в процессе: тот же пул и та же база данных, на часах теста

Крейт реализует трейты брокера из ядра поверх sqlx. Рантайм над ним один для всех брокеров. Всё, что относится к SQL, лежит в этом крейте: дескриптор InboxQueue, цикл захвата, три формы захвата, то, что записывает каждый исход, и SQL каждой базы данных. Крейт подписывается по дескриптору и по имени, берёт пакеты, передаёт обработчику в режиме строки саму строку, отдаёт ключ партиционирования диспетчеризации по ключу, публикует в таблицы и описывает свой сервер в генерируемом документе. Запроса-ответа и транзакционного издателя в нём нет: транзакционный режим - это шаг подписки, на котором обработчик пишет через транзакцию своей доставки.

Параметры пула принадлежат сервису: его размер, его настройки и его миграции. SqlxBroker::from_url записывает их без ввода-вывода, поэтому приложение строится ещё до запуска рантайма; брокер строит пул при подключении и закрывает его при завершении работы. Сервис, у которого один пул на брокер и на собственный код, строит его в своём main, внутри рантайма, и передаёт в SqlxBroker::new, который его никогда не закрывает.

SQL

Каждый оператор подписки строит диалект. Встроены диалекты PostgreSQL, MySQL и MariaDB, а также SQLite. AnyPool берёт диалект той базы данных, к которой обращается при подключении брокера, а сервис может подключить собственный диалект для другой базы данных. Подписка строит свои операторы один раз, при открытии, по описанию таблицы - значению const, которое выдаёт derive или написанный вручную InboxSpec. Поэтому сообщение стоит одинаково с derive, с ручным описанием и с собственным диалектом сервиса. Затем старт подготавливает каждый оператор на сервере, так что отсутствующая таблица или столбец останавливает сервис до первого сообщения.

Публикация

Переобработка не покидает строку. Дескриптор объявляет, что база данных перемещает доставку сама, поэтому retry_after - это UPDATE столбца retry_after строки, и копия никогда не публикуется: это показывает страница Завершение. Публикация записывает строку через собственную реализацию Publish типа строки. Repository называет свою таблицу во время компиляции. Routed - издатель подключённого брокера по умолчанию, поэтому через него идут ответы; он ищет имя сообщения среди маршрутов, с которыми построен брокер. Оба принимают закодированные байты как заимствованный срез, поэтому буфер ответа переиспользуется от сообщения к сообщению.

То, что остаётся на каждом сообщении, крейт указывает сам, а страница Бенчмарки это измеряет:

• Routed делает один поиск в хеш-таблице и один динамический вызов в Publish строки. Future этого вызова хранится на месте, в слоте на 1024 байта, и попадает в кучу, только когда он больше.
• Публикация читает флаг закрытия брокера, одна атомарная загрузка, и будит подписки своей таблицы в процессе, по одной атомарной операции на каждую.
• Обработчик, который читает Tx или Ctx<keys::Pool>, платит одним увеличением счётчика ссылок на доставку.
• Подписка по имени над строкой, события которой - собственные события сервиса, платит одной доставкой в куче и одним future завершения в куче на сообщение.

Режим в процессе

С фичей testing TestApp::start подключает брокер в процессе. Ниже пула ничего не подменяется: брокер в процессе обращается к базе данных, которую называет тест, выполняет SQL сервиса, его схему и его формы захвата и возвращает ошибку там же, где база данных. SQLite в памяти обслуживает тест без сервера. Меняется время. Брокер читает «сейчас» по часам tokio, привязанным к настенным часам в момент подключения, поэтому тест на остановленных часах двигает время очереди через tb.advance(..): строка с retry_after становится готовой к захвату, когда тест переводит часы дальше её срока. Каждое обращение к базе данных выполняется вне этих часов, поэтому остановленные часы не перескакивают к следующему таймеру, пока ответ базы данных ещё в пути. Часы теста заменяют только SystemClock: таблица на DatabaseClock читает часы базы данных, а собственные часы сервиса продолжают действовать, как в рабочей среде.

Обвязка дожидается работы базы данных по учёту брокера. Она засчитывает сообщение, когда оно попадает в очередь, которую читает подписка теста, и ещё раз, когда переобработка возвращает его строку, поэтому tb.settle() возвращается, когда каждая строка, попавшая в очередь, доставлена и завершена.

Как части запускаются и останавливаются, показывает страница Жизненный цикл, а путь одной строки - страница Путь.