Подключение к промышленному серверу

Управляемый сервис MQTT, например AWS IoT Core, HiveMQ Cloud или EMQX Cloud, требует учётных данных и TLS, часто с клиентским сертификатом. Это руководство настраивает брокер для такого сервиса по окружению развёртывания.

Учётные данные и TLS

src/main.rs

/// Where the server is and who the service is, as the deployment provides them.
struct Deployment {
    url: String,
    client_id: String,
    username: String,
    password: String,
    ca: Vec<u8>,
    cert: Vec<u8>,
    key: Vec<u8>,
}

impl Deployment {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        let file = |name: &str| fs::read(read(name)).unwrap_or_default();
        Self {
            url: read("MQTT_URL"),
            client_id: read("MQTT_CLIENT_ID"),
            username: read("MQTT_USERNAME"),
            password: read("MQTT_PASSWORD"),
            ca: file("MQTT_CA"),
            cert: file("MQTT_CERT"),
            key: file("MQTT_KEY"),
        }
    }
}

const LARGEST_PACKET: NonZeroU32 = NonZeroU32::new(256 * 1024).unwrap();
const IN_FLIGHT: NonZeroU16 = NonZeroU16::new(200).unwrap();

fn broker(deployment: Deployment) -> MqttBroker {
    MqttBroker::new(deployment.url, deployment.client_id)
        .credentials(deployment.username, deployment.password)
        .tls_ca(deployment.ca)
        .tls_client_auth(deployment.cert, deployment.key)
        .keep_alive(Duration::from_secs(30))
        .max_packet_size(LARGEST_PACKET)
        .receive_maximum(IN_FLIGHT)
}

Все настройки развёртывание передаёт в переменных окружения, а сертификаты - файлами. credentials задаёт имя пользователя и пароль в пакете протокола CONNECT. Логин внутри URL, например mqtt://user:p%40ss@host, тоже работает, и его символы раскодируются из процентной записи; credentials имеет приоритет над ним. В генерируемый документ AsyncAPI они не попадают: он указывает только хост и порт, к которым подключается сервис.

mqtts:// включает TLS, на порту 8883, если URL не называет другой. tls_ca - удостоверяющий центр в формате PEM, по которому клиент проверяет сервер. Клиент доверяет этому центру и никакому другому хранилищу, поэтому URL mqtts:// без tls_ca не даёт сервису запуститься, и ошибка называет tls_ca. tls_client_auth добавляет сертификат и ключ клиента для сервиса, который проверяет клиентов по сертификату.

Свой идентификатор клиента у каждого процесса

Идентификатор клиента - второй аргумент MqttBroker::new, и по нему сервер находит сессию. Второе соединение с тем же идентификатором забирает сессию себе и закрывает первое. Поэтому у двух реплик сервиса идентификаторы никогда не совпадают: здесь каждая берёт свой из MQTT_CLIENT_ID, например имя пода. Реплики, которые делят работу, вместо этого входят в группу совместной подписки.

Ограничения

• keep_alive - как часто клиент подтверждает, что жив, на простаивающем соединении. Сервер закрывает соединение, которое молчит в полтора раза дольше. Минимум клиента rumqttc - 5 секунд, значение по умолчанию - 60.
• max_packet_size - наибольший пакет протокола, который принимает сервис. Пакет больше сервер ему не отправит. Значение брокера по умолчанию - 1 МиБ, больше собственного значения клиента, 10 КиБ.
• receive_maximum ограничивает число доставок с QoS 1 и 2 в обработке (Гарантии доставки).

Сервис

Макросы

src/main.rs

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("ingest"))]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
    HandlerOutcome::ack()
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(broker, |b| {
        b.include(record)
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker(Deployment::from_env()))
}

Вручную

src/main.rs

struct Record;

impl Handle<Telemetry> for Record {
    fn handle(
        &self,
        telemetry: &Telemetry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
        ready(Ok(()))
    }
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(broker, |b| {
        let fleet = MqttFilter::new("devices/+/telemetry")
            .qos(Qos::AtLeastOnce)
            .shared("ingest");
        b.include(
            subscriber("devices/+/telemetry", Record)
                .map_source(|_| fleet)
                .build(),
        )
        .out_retry(Publish::default())
        .to("devices/retry/telemetry");
    })
}

fn app() -> impl App<State = ()> {
    service(broker(Deployment::from_env()))
}

fn main() -> ExitCode {
    run_main(app)
}

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

Проверка конфигурации

Тесты запускают рабочий service с рабочим broker, которому даны тестовые настройки, через TestApp::start. Внутрипроцессный сервер читает конфигурацию брокера и отклоняет то, что отклонил бы клиент, не подключаясь к серверу:

• keep-alive меньше 5 секунд не даёт сервису запуститься, и сообщение об ошибке называет минимум клиента;
• URL mqtts:// без tls_ca не даёт сервису запуститься, и ошибка называет tls_ca.

Учётные данные и TLS-рукопожатие проверяет только настоящий сервер. Живые тесты страницы Security проверяют и то и другое на слушателях Mosquitto, которые этого требуют.