I2C detection works well
This commit is contained in:
Priec
230
mqtt_display/src/mqtt/client.rs
Normal file
230
mqtt_display/src/mqtt/client.rs
Normal file
@@ -0,0 +1,230 @@
|
||||
// src/mqtt/client.rs
|
||||
use embassy_futures::select::{select, Either};
|
||||
use embassy_net::{tcp::TcpSocket, Stack};
|
||||
use embassy_time::{Duration, Timer};
|
||||
use log::info;
|
||||
use rust_mqtt::client::client::MqttClient;
|
||||
use rust_mqtt::client::client_config::{ClientConfig, MqttVersion};
|
||||
use rust_mqtt::packet::v5::publish_packet::QualityOfService;
|
||||
use rust_mqtt::packet::v5::reason_codes::ReasonCode;
|
||||
use rust_mqtt::utils::rng_generator::CountingRng;
|
||||
use static_cell::ConstStaticCell;
|
||||
|
||||
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
|
||||
use embassy_sync::channel::{Channel, Receiver};
|
||||
|
||||
use heapless::{String as HString, Vec as HVec};
|
||||
|
||||
use crate::mqtt::config::mqtt_broker_endpoint;
|
||||
|
||||
const RECONNECT_DELAY_SECS: u64 = 5;
|
||||
const KEEPALIVE_SECS: u64 = 60;
|
||||
const PING_PERIOD: Duration = Duration::from_secs(KEEPALIVE_SECS / 2);
|
||||
|
||||
// Limits for small, static buffers (no heap)
|
||||
pub const TOPIC_MAX: usize = 128;
|
||||
pub const PAYLOAD_MAX: usize = 512;
|
||||
const COMMAND_QUEUE: usize = 8;
|
||||
const EVENT_QUEUE: usize = 8;
|
||||
|
||||
// TCP socket buffers (for embassy-net TcpSocket)
|
||||
static TCP_RX_BUFFER: ConstStaticCell<[u8; 2048]> = ConstStaticCell::new([0; 2048]);
|
||||
static TCP_TX_BUFFER: ConstStaticCell<[u8; 2048]> = ConstStaticCell::new([0; 2048]);
|
||||
|
||||
// MQTT client buffers (separate from the TcpSocket's buffers)
|
||||
static MQTT_TX_BUF: ConstStaticCell<[u8; 1024]> = ConstStaticCell::new([0; 1024]);
|
||||
static MQTT_RX_BUF: ConstStaticCell<[u8; 1024]> = ConstStaticCell::new([0; 1024]);
|
||||
|
||||
// Tie TcpSocket lifetime to session
|
||||
type Client<'a, 'net> = MqttClient<'a, TcpSocket<'net>, 8, CountingRng>;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct IncomingMsg {
|
||||
pub topic: HString<TOPIC_MAX>,
|
||||
pub payload: HVec<u8, PAYLOAD_MAX>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct PublishMsg {
|
||||
topic: HString<TOPIC_MAX>,
|
||||
payload: HVec<u8, PAYLOAD_MAX>,
|
||||
qos: QualityOfService,
|
||||
retain: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
enum Command {
|
||||
Publish(PublishMsg),
|
||||
Subscribe(HString<TOPIC_MAX>),
|
||||
}
|
||||
|
||||
static CMD_CHAN: Channel<CriticalSectionRawMutex, Command, COMMAND_QUEUE> = Channel::new();
|
||||
static EVT_CHAN: Channel<CriticalSectionRawMutex, IncomingMsg, EVENT_QUEUE> = Channel::new();
|
||||
|
||||
// Public API
|
||||
pub async fn mqtt_publish(topic: &str, payload: &[u8], qos: QualityOfService, retain: bool) {
|
||||
CMD_CHAN
|
||||
.send(Command::Publish(PublishMsg {
|
||||
topic: truncate_str::<TOPIC_MAX>(topic),
|
||||
payload: truncate_payload(payload),
|
||||
qos,
|
||||
retain,
|
||||
}))
|
||||
.await;
|
||||
}
|
||||
|
||||
pub async fn mqtt_subscribe(topic: &str) {
|
||||
CMD_CHAN.send(Command::Subscribe(truncate_str::<TOPIC_MAX>(topic))).await;
|
||||
}
|
||||
|
||||
pub fn mqtt_events(
|
||||
) -> Receiver<'static, CriticalSectionRawMutex, IncomingMsg, EVENT_QUEUE> {
|
||||
EVT_CHAN.receiver()
|
||||
}
|
||||
|
||||
// Helper functions for memory-safe truncation
|
||||
fn truncate_str<const N: usize>(s: &str) -> HString<N> {
|
||||
let mut h = HString::new();
|
||||
let _ = h.push_str(&s[..s.len().min(N)]);
|
||||
h
|
||||
}
|
||||
|
||||
fn truncate_payload(data: &[u8]) -> HVec<u8, PAYLOAD_MAX> {
|
||||
let mut v = HVec::new();
|
||||
let _ = v.extend_from_slice(&data[..data.len().min(PAYLOAD_MAX)]);
|
||||
v
|
||||
}
|
||||
|
||||
// MQTT configuration and client setup
|
||||
fn build_client_config() -> ClientConfig<'static, 8, CountingRng> {
|
||||
let mut cfg = ClientConfig::new(MqttVersion::MQTTv5, CountingRng(0));
|
||||
cfg.keep_alive = KEEPALIVE_SECS as u16;
|
||||
cfg.add_client_id("esp32-client");
|
||||
cfg
|
||||
}
|
||||
|
||||
fn build_client<'a, 'net>(
|
||||
socket: TcpSocket<'net>,
|
||||
mqtt_tx: &'a mut [u8],
|
||||
mqtt_rx: &'a mut [u8],
|
||||
) -> Client<'a, 'net> {
|
||||
let mqtt_tx_len = mqtt_tx.len();
|
||||
let mqtt_rx_len = mqtt_rx.len();
|
||||
MqttClient::new(socket, mqtt_tx, mqtt_tx_len, mqtt_rx, mqtt_rx_len, build_client_config())
|
||||
}
|
||||
|
||||
// Connection lifecycle and main session loop
|
||||
|
||||
async fn connect_tcp<'net>(socket: &mut TcpSocket<'net>) -> Result<(), ()> {
|
||||
match socket.connect(mqtt_broker_endpoint()).await {
|
||||
Ok(_) => {
|
||||
info!("Connected TCP to MQTT broker");
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
info!("TCP connect failed: {:?}", e);
|
||||
Err(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn connect_mqtt(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> {
|
||||
client.connect_to_broker().await
|
||||
}
|
||||
|
||||
async fn run_loop(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> {
|
||||
let default_topic = "esp32/topic";
|
||||
match client.subscribe_to_topic(default_topic).await {
|
||||
Ok(_) => info!("Subscribed to '{}'", default_topic),
|
||||
Err(e) => info!("Default subscribe failed: {:?}", e),
|
||||
};
|
||||
|
||||
loop {
|
||||
let net_or_ping = select(client.receive_message(), Timer::after(PING_PERIOD));
|
||||
|
||||
match select(CMD_CHAN.receive(), net_or_ping).await {
|
||||
Either::First(cmd) => handle_command(client, cmd).await?,
|
||||
Either::Second(Either::First(result)) => handle_incoming(result).await?,
|
||||
Either::Second(Either::Second(_)) => client.send_ping().await?,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_command(client: &mut Client<'_, '_>, cmd: Command) -> Result<(), ReasonCode> {
|
||||
match cmd {
|
||||
Command::Publish(msg) => {
|
||||
client
|
||||
.send_message(msg.topic.as_str(), &msg.payload, msg.qos, msg.retain)
|
||||
.await
|
||||
}
|
||||
Command::Subscribe(topic) => {
|
||||
client.subscribe_to_topic(topic.as_str()).await?;
|
||||
info!("Subscribed to '{}'", topic);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_incoming(result: Result<(&str, &[u8]), ReasonCode>) -> Result<(), ReasonCode> {
|
||||
let (topic, payload) = result?;
|
||||
EVT_CHAN
|
||||
.send(IncomingMsg {
|
||||
topic: truncate_str::<TOPIC_MAX>(topic),
|
||||
payload: truncate_payload(payload),
|
||||
})
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Session and reconnect control
|
||||
async fn run_one_session(
|
||||
stack: Stack<'static>,
|
||||
tcp_rx: &mut [u8],
|
||||
tcp_tx: &mut [u8],
|
||||
mqtt_tx: &mut [u8],
|
||||
mqtt_rx: &mut [u8],
|
||||
) -> Result<(), ()> {
|
||||
let mut socket = TcpSocket::new(stack, tcp_rx, tcp_tx);
|
||||
if connect_tcp(&mut socket).await.is_err() {
|
||||
return Err(());
|
||||
}
|
||||
|
||||
let mut client = build_client(socket, mqtt_tx, mqtt_rx);
|
||||
match connect_mqtt(&mut client).await {
|
||||
Ok(_) => info!("MQTT CONNACK received"),
|
||||
Err(reason) => {
|
||||
info!("MQTT connect failed: {:?}", reason);
|
||||
return Err(());
|
||||
}
|
||||
}
|
||||
|
||||
run_loop(&mut client).await.map_err(|_| ())
|
||||
}
|
||||
|
||||
// Main MQTT embassy task
|
||||
#[embassy_executor::task]
|
||||
pub async fn mqtt_task(stack: Stack<'static>) {
|
||||
info!("MQTT task starting...");
|
||||
|
||||
let tcp_rx = TCP_RX_BUFFER.take();
|
||||
let tcp_tx = TCP_TX_BUFFER.take();
|
||||
let mqtt_tx = MQTT_TX_BUF.take();
|
||||
let mqtt_rx = MQTT_RX_BUF.take();
|
||||
|
||||
loop {
|
||||
let _ = run_one_session(
|
||||
stack,
|
||||
&mut tcp_rx[..],
|
||||
&mut tcp_tx[..],
|
||||
&mut mqtt_tx[..],
|
||||
&mut mqtt_rx[..],
|
||||
)
|
||||
.await;
|
||||
|
||||
info!(
|
||||
"Reconnecting in {}s after session end/failure",
|
||||
RECONNECT_DELAY_SECS
|
||||
);
|
||||
Timer::after(Duration::from_secs(RECONNECT_DELAY_SECS)).await;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user