Демультиплексирование

Одно соединение обслуживает все подписки: идентификаторы подписок различают пересекающиеся фильтры, а подписки на один фильтр получают доставки по очереди.

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

1. Фильтры на соединении: devices/+/telemetry идентификатор 17; берёт по очереди; берёт следующий; devices/# идентификатор 42; одна подписка берёт все
2. PUBLISH devices/dev42/telemetry идентификатор 17 → devices/+/telemetry идентификатор 17
3. devices/+/telemetry идентификатор 17 → берёт по очереди
4. devices/+/telemetry идентификатор 17 → берёт следующий: далее
5. PUBLISH devices/dev42/telemetry идентификатор 42 → devices/# идентификатор 42
6. devices/# идентификатор 42 → одна подписка берёт все
7. Пакет, который не взяла ни одна открытая подписка, ждёт подписку, которая откроется позже; таких не больше 1024

Сервер подписывает фильтр один раз на соединение. Два обработчика на одном фильтре или два участника группы совместной подписки в одном сервисе - это одна подписка на сервере и две подписки в сервисе. Крейт передаёт им доставки фильтра по очереди. Фильтр остаётся подписанным, пока открыта хотя бы одна из них, а остановка сервиса оставляет его в сессии. Он подписан на наибольшем уровне качества обслуживания, который запрашивает любая из них.

Разные фильтры, которые могут подойти к одной теме, например devices/+/telemetry рядом с devices/# или группа совместной подписки рядом с обычной подпиской на её фильтр, - отдельные подписки на сервере. Для сообщения, к которому подходят оба, Mosquitto отправляет по пакету протокола на каждый. MQTT 5 позволяет подписке иметь идентификатор, и сервер указывает его в каждом пакете, который отправляет для этой подписки. Крейт даёт идентификатор только фильтру, который пересекается с другим фильтром на соединении, и направляет каждый пакет по указанному в нём идентификатору.

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

Идентификаторы подписок для сервера необязательны, и его CONNACK сообщает, поддерживает ли он их. На сервере без них открытие фильтра, который пересекается с другим фильтром на том же соединении, возвращает ошибку, которая называет оба фильтра, и ничего не подписывается. Такие фильтры можно развести по двум брокерам с разными идентификаторами клиентов. Если сервер отправляет один пакет протокола для двух фильтров, подтверждение отправляет доставка первому фильтру, а ack доставки второму возвращает AckError::Unsupported.

Пакет, который не взяла ни одна открытая подписка, придерживается, потому что возобновлённая сессия отправляет накопленные сообщения раньше, чем сервис откроет подписки, к которым они относятся. Ждут до 1024 пакетов, а сверх этого самый старый отбрасывается с предупреждением.

Партиции воркеров

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

1. Подписка тема или фильтр на общем соединении → Поток MqttSubscriber по одной доставке: доставлено
2. Поток MqttSubscriber по одной доставке → workers(2) партиции воркеров берут доставки по очереди: по одной
3. workers(2) партиции воркеров берут доставки по очереди → Партиция воркеров 1 свои доставки по порядку
4. workers(2) партиции воркеров берут доставки по очереди → Партиция воркеров 2 свои доставки по порядку
5. by_key: одно значение partition-key - одна партиция; без заголовка - следующая по кругу
6. Между процессами: группа совместной подписки отдаёт каждое сообщение одному участнику

Без workers один цикл вызывает обработчик для каждой доставки подписки по очереди. Порядок, в котором их отправил сервер, сохраняется, и ничего не выполняется параллельно.

workers(n) запускает n партиций воркеров, и каждая выполняет свои доставки по порядку. Партиции берут доставки по очереди, поэтому доставки одной темы выполняются параллельно в разных партициях. При workers(n, by_key) партицию выбирает заголовок partition-key, который поставил издатель: доставки с одним ключом выполняются по порядку, а доставка без заголовка попадает в следующую партицию по кругу.

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

Обработчик пакетов берёт до того размера, который задан в точке монтирования. В одном PUBLISH одно сообщение, поэтому крейт собирает пакет на клиенте. Пакет закрывается, когда набирает этот размер или через 20 миллисекунд после первой доставки.