Жизненный цикл

MqttBroker::new записывает конфигурацию, подключение ждёт CONNACK от сервера, запуск подписывает каждый фильтр, а остановка отправляет DISCONNECT.

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

1. Брокер: MqttBroker::new (url, client_id); ConnectedMqttBroker; Closed: () сообщать нечего
2. В MQTT: URL, client id и настройки, без I/O; Проверка настроек: порт, keep-alive, TLS; Запуск задачи, ожидание CONNACK 30 с; Каждая подписка: SUBSCRIBE, SUBACK; Переподключение с паузой, переподписка; Пометка о закрытии, DISCONNECT, сброс
3. MqttBroker::new (url, client_id) → ConnectedMqttBroker: connect(self)
4. ConnectedMqttBroker → Closed: () сообщать нечего: shutdown(self)
5. URL, client id и настройки, без I/O → Проверка настроек: порт, keep-alive, TLS
6. Проверка настроек: порт, keep-alive, TLS → Запуск задачи, ожидание CONNACK 30 с
7. Запуск задачи, ожидание CONNACK 30 с → Каждая подписка: SUBSCRIBE, SUBACK
8. Каждая подписка: SUBSCRIBE, SUBACK → Переподключение с паузой, переподписка
9. Переподключение с паузой, переподписка → Пометка о закрытии, DISCONNECT, сброс
10. Издатель, переживший соединение, отвечает MqttError::NotConnected, а не молчаливым успехом

new не выполняет ввода-вывода, поэтому билдер сервиса остаётся синхронным. Он принимает URL, mqtt://host:port или mqtts://host:port, и идентификатор клиента, а рядом с ними записываются настройки билдера. Рантайм подключает брокер, когда запускается приложение.

Подключение сначала проверяет настройки, до всякого ввода-вывода. Порт, который не является числом, keep-alive меньше 5 секунд, минимума клиента, и TLS без удостоверяющего центра, которому можно доверять, прерывают запуск с ошибкой MqttError::Invalid, которая называет настройку.

Затем подключение запускает задачу соединения и до 30 секунд ждёт от сервера первый CONNACK. То, что может пройти со временем, например сервер, который ещё не поднялся, задача пробует снова. Отказ, который соединение не преодолеет повторами, сразу заканчивает ожидание с MqttError::Connect: неверные учётные данные, идентификатор клиента, который отклоняет сервер, узел, который не поддерживает MQTT, или рукопожатие TLS, которое не проходит с этими сертификатами. Если время ожидания вышло, подключение возвращает последнюю ошибку соединения, которую видела задача.

Дескриптор каждой подписки разрешается один раз, на подключённом брокере. Он отправляет SUBSCRIBE и ждёт SUBACK, поэтому подписка уже есть на сервере, когда приложение сообщает о запуске. Фильтр, который отклоняет сервер, прерывает запуск с MqttError::Subscribe.

Пока сервис работает, задача переподключает потерянное соединение. Перед первой попыткой она ждёт 100 миллисекунд и удваивает ожидание до 5 секунд. Если CONNACK сообщает, что сессии больше нет, задача заново подписывает каждый фильтр. Отказ, который соединение не преодолеет повторами, вместо этого завершает задачу, и каждая подписка возвращает его как MqttError::Receive.

Остановка сначала помечает соединение закрытым, чтобы издатель, который делит с ним соединение, уже не смог отправить сообщение. Затем она ставит DISCONNECT в очередь после всех подтверждений, которые уже в ней стоят, и до 5 секунд ждёт, пока задача их все запишет. Сервер видит штатное окончание соединения и не публикует последнюю волю. Больше остановка ничего не возвращает: закрытое состояние не содержит диагностики.

Издатель, которого инстанцировали до остановки, - дескриптор, который остановка не может забрать. Он возвращает MqttError::NotConnected. Доставка, подтверждённая после остановки, возвращает ошибку брокера.

Внутрипроцессный транспорт проходит те же переходы. Его подключение проверяет настройки так же, как connect, поэтому конфигурацию, которую отклоняет путь через сервер, отклоняет и тест.