Устройство

Где стоит ruststream-rumqttc: между рантаймом RustStream и клиентом rumqttc, на одном соединении, которое обслуживает собственная задача.

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

1. Ваш сервис: #[subscriber] обработчики-подписчики; RustStream::new(..) .with_broker(..)
2. Ядро RustStream: Рантайм: диспетчер, пулы, завершение, переобработка
3. ruststream-rumqttc: MqttSubscriber выдаёт MqttMessage; MqttPublisher из политики; Контекст: DeliveryTopic; Транспорт в процессе; MqttBroker одно соединение; MqttTopic, MqttFilter; Задача соединения: раздаёт PUBLISH
4. rumqttc: EventLoop читает и пишет; AsyncClient принимает запросы
5. #[subscriber] обработчики-подписчики → Рантайм: диспетчер, пулы, завершение, переобработка: монтируются
6. Рантайм: диспетчер, пулы, завершение, переобработка → MqttSubscriber выдаёт MqttMessage: опрашивает поток
7. Задача соединения: раздаёт PUBLISH → MqttSubscriber выдаёт MqttMessage
8. Задача соединения: раздаёт PUBLISH → EventLoop читает и пишет: опрашивает
9. MqttPublisher из политики → AsyncClient принимает запросы
10. AsyncClient принимает запросы → EventLoop читает и пишет
11. EventLoop читает и пишет → Сервер MQTT

Крейт реализует трейты брокера из ядра поверх rumqttc и работает только по MQTT 5. Рантайм над ним один для всех брокеров. Всё, что относится к MQTT, лежит в этом крейте: дескрипторы, что отправляет ack, куда уходит копия для переобработки, тема доставки и отображение заголовков на свойства MQTT 5. Заголовки передаются собственными свойствами протокола, поэтому клиент на любом языке программирования читает обычное сообщение MQTT.

rumqttc делит соединение на две части. AsyncClient - дескриптор, все копии которого ставят запросы в одну очередь: подписку, публикацию, подтверждение. EventLoop пишет эти запросы в сокет и читает то, что отправляет сервер, и делает то и другое, только пока его кто-то опрашивает. Опрос же отправляет пинги keep-alive и продвигает подтверждения.

Подключённый брокер держит одну такую пару. Крейт опрашивает цикл событий в одной собственной задаче, которую запускает при подключении, и эта задача только опрашивает и разводит пакеты протокола. Поэтому медленный обработчик никогда не останавливает соединение: пинги, подтверждения и доставки других подписок продолжают идти. Подписка, публикация и подтверждение идут через AsyncClient из той задачи, которая делает вызов.

Задача соединения передаёт каждый PUBLISH, который отправляет сервер, подпискам, к которым он относится, по одному каналу на подписку. Все подписки и издатели брокера делят одно соединение, поэтому сервис с множеством обработчиков держит на сервере одну сессию. Как одно соединение различает свои подписки, показывает страница Демультиплексирование.

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

Внутрипроцессный транспорт

С фичей testing TestApp::start подключает тот же брокер к внутрипроцессному транспорту, а не к серверу. Дескрипторы, политики и доставки остаются теми же, что в рабочей сборке, а транспорт читает собственные настройки брокера. Публикация проходит ту же проверку и то же отображение заголовков на свойства, что и при передаче по сети.

Транспорт моделирует сервер, с которым работает это соединение. Он сопоставляет фильтры с + и # и соблюдает правило $. Он отправляет по одному пакету протокола на каждую подходящую подписку, на меньшем уровне качества обслуживания и с идентификатором подписки, и каждый пакет проходит то же демультиплексирование, что выполняет задача соединения. Поэтому группа совместной подписки получает одну копию на всех участников, а пересекающиеся фильтры получают каждый свою. Транспорт хранит сохранённое сообщение каждой темы, отбрасывает сообщение крупнее max_packet_size и отклоняет подтверждение после остановки.

То, что принадлежит серверу и второму соединению, проверяет только живой прогон: постоянная сессия и повторная доставка при её возобновлении, последняя воля, окно Receive Maximum, обмен подтверждениями QoS 1 и QoS 2, управление доступом и несколько соединений на одном сервере. Каждый брокер, подключённый внутри процесса, - отдельный сервер.

Без фичи транспорт не компилируется. Дескриптор соединения в рабочей сборке занимает ровно столько же, сколько AsyncClient, который он оборачивает, и компилятор проверяет этот размер.