Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Публикация данных в брокер MQTT

В этой главе мы заменяем временный HTTP-запрос из предыдущей главы на MQTT. Плата подключается к брокеру MQTT, считывает датчик SHTC3 и публикует температуру раз в секунду. Готовый код для этой главы находится в project/part3/.

Инициализация I2C, Wi-Fi и сетевого стека из главы Подключение к Wi-Fi остаётся прежней. Основные изменения — клиент MQTT, отдельная задача, которая владеет датчиком, и локальный брокер, позволяющий просматривать каждую публикацию.

Учётные данные Wi-Fi

В этой главе по-прежнему используются SSID и PASSWORD, заданные на этапе компиляции, как и в предыдущей главе. Оставив настройку Wi-Fi без изменений, мы можем сосредоточиться на MQTT. В следующей главе эти учётные данные будут заменены порталом Wi-Fi-провизионинга.

Что добавляет MQTT

MQTT — протокол «публикация/подписка» (publish/subscribe). Вместо того чтобы отправлять HTTP-запрос напрямую определённой конечной точке приложения, клиент публикует полезную нагрузку в именованный топик на брокере. Другие клиенты могут подписаться на этот топик и получать эту нагрузку.

В этой главе используются три понятия MQTT:

  • Брокер принимает подключения и маршрутизирует сообщения.
  • ESP32-C3 является издателем (publisher).
  • measurement/temperature — топик, в который устройство публикует данные.

Локальный брокер, который мы запустим позже в этой главе, также подписывается на measurement/#. Символ # соответствует всем топикам внутри measurement, поэтому логгер брокера выводит сообщения, отправленные в measurement/temperature.

Добавление клиента MQTT

Реализация MQTT находится в новом модуле:

mod mqtt;

Мы используем крейт rust-mqtt:

rust-mqtt = { version = "0.5.1", default-features = false, features = ["bump", "v5"] }

Feature-флаги по умолчанию отключены, потому что это приложение no_std. Feature v5 включает MQTT версии 5, которая соответствует локальному брокеру, а bump включает хранение в bump-буфере крейта. Bump-буфер заранее резервирует блок памяти фиксированного размера и позволяет клиенту использовать его для пакетов MQTT без отдельного выделения кучи для каждой операции.

Запуск задачи MQTT

В предыдущей главе main владел датчиком и выполнял каждое чтение и каждый HTTP-запрос. Теперь мы переносим этот цикл в задачу Embassy:

spawner.spawn(mqtt_task(stack, sht).expect("failed to spawn MQTT task"));

Stack — лёгкий копируемый дескриптор сетевого стека, поэтому его можно передать задаче, пока net_task продолжает опрашивать Runner стека. Драйвер SHTC3 передаётся в mqtt_task, что даёт этой задаче эксклюзивное владение датчиком I2C.

После запуска фоновых задач у main больше нет работы приложения. Он остаётся активным и уступает управление исполнителю:

// Keep main task alive
loop {
    embassy_time::Timer::after(embassy_time::Duration::from_secs(60)).await;
}

Задача подключения сети, Runner сети и задача MQTT теперь могут продвигаться независимо друг от друга, когда другие задачи ожидают.

Инициализация MQTT

Задача MQTT сначала резервирует буферы приёма и передачи для своего TCP-сокета:

let mut rx_buffer = [0; 4096];
let mut tx_buffer = [0; 4096];

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

Ожидание сети

MQTT работает поверх TCP, поэтому задача не может подключиться к брокеру, пока Wi-Fi не подключён и DHCP не настроил сетевой стек:

// Wait for network to be ready before attempting connection
debug!("Waiting for WiFi link to come up...");
stack.wait_link_up().await;
debug!("WiFi link up, waiting for network configuration...");

// Wait for DHCP to assign an IP address
stack.wait_config_up().await;

debug!("Waiting to get IP address...");
loop {
    if let Some(config) = stack.config_v4() {
        debug!("Got IP: {}", config.address);
        break;
    }
    Timer::after(Duration::from_millis(500)).await;
}

// Check if we still have a valid network config before proceeding
if !stack.is_config_up() {
    debug!("Network config lost, retrying...");
    continue;
}

wait_link_up() ждёт, пока поднимется канал Wi-Fi. wait_config_up() ждёт сетевой конфигурации, а config_v4() подтверждает, что IPv4-адрес доступен. Если конфигурация пропадёт до начала подключения к брокеру, внешний цикл начнёт последовательность заново.

Поиск брокера

Адрес и порт брокера считываются во время компиляции:

let host = match HOST_IP {
    Some(h) => h,
    None => {
        error!(
            "No HOST_IP set. Provide e.g. HOST_IP=10.0.0.10 (or hostname) and optional BROKER_PORT."
        );
        Timer::after(Duration::from_secs(5)).await;
        continue;
    }
};

// Default to rumqttd's v5 listener port (1884) unless overridden
let port: u16 = BROKER_PORT
    .and_then(|p| p.parse::<u16>().ok())
    .unwrap_or(1884);

Оба значения встраиваются в прошивку при её сборке. HOST_IP обязателен для логики приложения, но использует option_env!, поэтому отсутствие значения приводит к ошибке в логе, а не к ошибке компиляции. BROKER_PORT необязателен и по умолчанию равен 1884 — порту, который использует локальный брокер MQTT этого курса.

Несмотря на название, HOST_IP может содержать и имя хоста. Задача сначала пытается разобрать значение как IPv4-адрес. Если разбор не удаётся, она запрашивает у сетевого стека разрешение IPv4-адреса через DNS:

// If host is an IPv4 literal, bypass DNS
let address = match host.parse::<Ipv4Address>() {
    Ok(ipv4) => IpAddress::Ipv4(ipv4),
    Err(_) => match stack.dns_query(host, DnsQueryType::A).await {
        Ok(addresses) if !addresses.is_empty() => addresses[0],
        Ok(_) => {
            error!("DNS query returned no addresses for {}", host);
            Timer::after(Duration::from_secs(5)).await;
            continue;
        }
        Err(e) => {
            error!("DNS lookup error: {e:?}");
            Timer::after(Duration::from_secs(5)).await;
            continue;
        }
    },
};

Использование литерального IP-адреса позволяет избежать DNS-запроса. Это удобно для локального брокера, адрес которого обычно совпадает с IP-адресом компьютера, на котором вы работаете.

Открытие TCP и подключение MQTT

Далее задача открывает TCP-сокет к найденному адресу брокера:

let mut socket = TcpSocket::new(stack, &mut rx_buffer, &mut tx_buffer);
socket.set_timeout(Some(embassy_time::Duration::from_secs(10)));

let remote_endpoint = (address, port);
info!("connecting to MQTT broker at {}:{}...", host, port);
let connection = socket.connect(remote_endpoint).await;
if let Err(e) = connection {
    error!("connect error: {:?}", e);
    Timer::after(Duration::from_secs(5)).await;
    continue;
}
info!("connected!");

Десятисекундный таймаут не даёт попытке подключения ждать бесконечно. Если TCP-подключение не удалось, задача ждёт пять секунд и возвращается к началу внешнего цикла.

Когда TCP подключён, можно инициализировать клиент MQTT:

let mut mqtt_buffer_storage = [0; 1024];
let mut mqtt_buffer = BumpBuffer::new(&mut mqtt_buffer_storage);
let mut client = Client::<_, _, 1, 1, 1, 1>::new(&mut mqtt_buffer);
let connect_options = ConnectOptions::new().clean_start();
let client_id = MqttString::from_str("esp32c3").expect("valid MQTT client id");

if let Err(e) = client
    .connect(socket, &connect_options, Some(client_id))
    .await
{
    error!("MQTT connect error: {:?}", e);
    continue;
}

BumpBuffer размером 1 КиБ предоставляет рабочую память для пакетов MQTT. ConnectOptions::new().clean_start() просит брокер начать работу без предыдущей сессии MQTT, а esp32c3 идентифицирует это устройство для брокера. client.connect(...) отправляет пакет MQTT CONNECT через TCP-сокет и ждёт, пока брокер его примет.

TCP и MQTT — разные уровни соединения. Успешный вызов socket.connect(...) лишь устанавливает байтовый поток к брокеру. Успешный вызов client.connect(...) завершает рукопожатие MQTT поверх этого потока.

Наконец, задача создаёт топик и параметры публикации один раз для каждого соединения MQTT:

let topic = TopicName::new(
    MqttString::from_str("measurement/temperature").expect("valid MQTT topic string"),
)
.expect("valid MQTT topic name");
let publish_options = PublicationOptions::new(TopicReference::Name(topic)).retain();

Топик — measurement/temperature. Вызов retain() просит брокер сохранить последнее значение, поэтому новый подписчик получит актуальную температуру, не дожидаясь следующего показания датчика.

Публикация измерений

Внутренний цикл — основная рабочая часть задачи. Перед чтением датчика он проверяет, что канал Wi-Fi и сетевая конфигурация всё ещё доступны:

// Check network state before attempting operations
if !stack.is_link_up() || !stack.is_config_up() {
    debug!("Network connection lost, reconnecting...");
    break;
}

Если одно из них потеряно, break выходит из цикла публикации. Внешний цикл затем ждёт сеть и создаёт новые соединения TCP и MQTT.

При каждой публикации задача считывает датчик и форматирует температуру:

// Read sensor
let (temp, _) = match read_sensor(&mut sht).await {
    Some(reading) => reading,
    None => {
        Timer::after(Duration::from_secs(1)).await;
        continue;
    }
};

let mut temperature_string: heapless::String<32> = heapless::String::new();
write!(temperature_string, "{:.2}", temp).expect("write! failed!");

read_sensor по-прежнему возвращает и температуру, и влажность, но в этой главе публикуется только температура. Шаблон _ намеренно отбрасывает значение влажности. heapless::String фиксированной ёмкости хранит отформатированное число, что позволяет не выделять память в куче для каждого показания.

Байты строки становятся полезной нагрузкой MQTT:

if let Err(e) = client
    .publish(&publish_options, Bytes::from(temperature_string.as_bytes()))
    .await
{
    error!("MQTT publish error: {:?}", e);
    break;
}

Если публикация прошла успешно, задача ждёт одну секунду и снова считывает датчик. Если нет, задача выходит из внутреннего цикла и переподключается. Так у приложения есть единый путь восстановления после потери канала Wi-Fi, потери IP-конфигурации, закрытия TCP-сокета или ошибки MQTT.

Запуск приложения

Запустите брокер MQTT из корня репозитория:

cargo xtask mqtt-server

Команда запускает брокер MQTT версии 5 на порту 1884, подписывает свой логгер на measurement/# и выводит IP-адрес компьютера, если может его определить:

MQTT server listening on 0.0.0.0:1884
Host IP (best effort): 192.168.1.10
Remote devices should connect to 192.168.1.10:1884
Subscribed to `measurement/#` and printing all incoming MQTT messages.

Оставьте эту команду запущенной. В другом терминале перейдите в project/part3/ и прошейте приложение:

SSID="your-ssid" PASSWORD="your-password" HOST_IP="your-ip" cargo run --release

Замените your-ip на IP-адрес хоста, который вывела предыдущая команда, и укажите учётные данные своей сети. Плата и компьютер с брокером должны быть доступны друг для друга по сети. Чтобы использовать другой порт брокера, передайте один и тот же порт обеим командам:

cargo xtask mqtt-server --port 1885

SSID="your-ssid" PASSWORD="your-password" HOST_IP="your-ip" BROKER_PORT="1885" cargo run --release

После того как Wi-Fi, TCP и MQTT подключатся, в мониторе последовательного порта должны появиться показания датчика:

INFO - Wifi connected!
INFO - connecting to MQTT broker at 192.168.1.10:1884...
INFO - connected!
INFO -   24.31 °C | 42.18 %RH

Терминал брокера должен вывести каждую сохранённую публикацию:

Topic: measurement/temperature | Payload: 24.31

На этом этапе плата может поддерживать соединение MQTT и публиковать измерения датчика в брокер. В следующей главе мы заменим учётные данные Wi-Fi, заданные на этапе компиляции, порталом провизионинга.