diff --git a/final/src/bin/main.rs b/final/src/bin/main.rs index df163a5..8be0879 100644 --- a/final/src/bin/main.rs +++ b/final/src/bin/main.rs @@ -1,3 +1,5 @@ +// src/bin/main.rs + #![no_std] #![no_main] #![deny( @@ -6,19 +8,22 @@ )] use embassy_executor::Spawner; +use embassy_futures::select::{select, Either}; use embassy_net::{Runner, StackResources}; use embassy_time::{Duration, Timer}; use esp_alloc as _; use esp_backtrace as _; use esp_hal::{clock::CpuClock, rng::Rng, timer::timg::TimerGroup}; use esp_wifi::{ - EspWifiController, init, wifi::{ClientConfiguration, Configuration, WifiController, WifiDevice, WifiEvent, WifiState}, + EspWifiController, }; use log::info; use rust_mqtt::packet::v5::publish_packet::QualityOfService; -use projekt_final::mqtt::client::{mqtt_task, mqtt_publish}; +use projekt_final::mqtt::client::{ + mqtt_events, mqtt_publish, mqtt_subscribe, mqtt_task, IncomingMsg, +}; use defmt_rtt as _; extern crate alloc; @@ -53,7 +58,8 @@ async fn main(spawner: Spawner) -> ! { init(timg0.timer0, rng.clone()).unwrap() ); - let (controller, interfaces) = esp_wifi::wifi::new(&esp_wifi_ctrl, peripherals.WIFI).unwrap(); + let (controller, interfaces) = + esp_wifi::wifi::new(&esp_wifi_ctrl, peripherals.WIFI).unwrap(); let wifi_interface = interfaces.sta; @@ -95,18 +101,43 @@ async fn main(spawner: Spawner) -> ! { spawner.spawn(mqtt_task(stack)).expect("failed to spawn MQTT task"); info!("MQTT task started"); - loop { - // TODO example - mqtt_publish( - "esp32/topic", - b"hello from main", - QualityOfService::QoS1, - false, - ) - .await; + mqtt_publish("esp32/topic", b"hello from ESP32 (init)", QualityOfService::QoS1, false).await; + info!("Sent initial MQTT message"); - // Avoid spamming, just an example cadence - Timer::after(Duration::from_secs(5)).await; + // Subscribe to your topics (can be done anytime; command is queued) + mqtt_subscribe("esp32/topic").await; + + // Get a receiver for incoming MQTT messages + let mqtt_rx = mqtt_events(); + + loop { + // Drive both: either process an MQTT message or publish periodically + match select(mqtt_rx.receive(), Timer::after(Duration::from_secs(5))).await + { + // Received inbound MQTT message (from broker) + Either::First(msg) => { + handle_incoming(msg); + } + // Time-based example publish + Either::Second(_) => { + // mqtt_publish( + // "esp32/topic", + // b"hello from main", + // QualityOfService::QoS1, + // false, + // ) + // .await; + } + } + } +} + +fn handle_incoming(msg: IncomingMsg) { + if let Ok(txt) = core::str::from_utf8(&msg.payload) { + info!("MAIN RX [{}]: {}", msg.topic.as_str(), txt); + info!("Received MQTT message -> topic: '{}', payload: '{}'", msg.topic.as_str(), txt); + } else { + info!("MAIN RX [{}]: {:?}", msg.topic.as_str(), msg.payload); } } diff --git a/final/src/mqtt/client.rs b/final/src/mqtt/client.rs index 12f3f99..ae81fbe 100644 --- a/final/src/mqtt/client.rs +++ b/final/src/mqtt/client.rs @@ -9,8 +9,10 @@ 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; +use embassy_sync::channel::{Channel, Receiver}; + use heapless::{String as HString, Vec as HVec}; use crate::mqtt::config::mqtt_broker_endpoint; @@ -19,10 +21,11 @@ const RECONNECT_DELAY_SECS: u64 = 5; const KEEPALIVE_SECS: u64 = 60; const PING_PERIOD: Duration = Duration::from_secs(KEEPALIVE_SECS / 2); -// Small owned buffers for data from main (no heap, no lifetimes) +// Limits for small, static buffers (no heap) pub const TOPIC_MAX: usize = 128; pub const PAYLOAD_MAX: usize = 512; -const PUBLISH_QUEUE: usize = 4; +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]); @@ -32,10 +35,10 @@ static TCP_TX_BUFFER: ConstStaticCell<[u8; 2048]> = ConstStaticCell::new([0; 204 static MQTT_TX_BUF: ConstStaticCell<[u8; 1024]> = ConstStaticCell::new([0; 1024]); static MQTT_RX_BUF: ConstStaticCell<[u8; 1024]> = ConstStaticCell::new([0; 1024]); -// NOTE: Tie the TcpSocket lifetime to the session (not 'static), and -// the MQTT buffers lifetime to 'a. +// Tie TcpSocket lifetime to session, MQTT buffers to 'a (fixes E0521). type Client<'a, 'net> = MqttClient<'a, TcpSocket<'net>, 8, CountingRng>; +// ===== Commands from main to MQTT task ===== #[derive(Clone)] struct PublishMsg { topic: HString, @@ -44,10 +47,28 @@ struct PublishMsg { retain: bool, } -static PUB_CHAN: Channel = +#[derive(Clone)] +enum MqttCommand { + Publish(PublishMsg), + Subscribe(HString), +} + +static CMD_CHAN: Channel = Channel::new(); -// Public API for main: enqueue a publish +// ===== Events (messages) from MQTT task to main ===== +#[derive(Clone)] +pub struct IncomingMsg { + pub topic: HString, + pub payload: HVec, +} + +static EVT_CHAN: Channel = + Channel::new(); + +// === Public API for main === + +// Enqueue a publish (non-alloc, copies into small heapless buffers) pub async fn mqtt_publish( topic: &str, payload: &[u8], @@ -59,16 +80,30 @@ pub async fn mqtt_publish( let mut p: HVec = HVec::new(); let take = core::cmp::min(payload.len(), PAYLOAD_MAX); let _ = p.extend_from_slice(&payload[..take]); - PUB_CHAN - .send(PublishMsg { + + CMD_CHAN + .send(MqttCommand::Publish(PublishMsg { topic: t, payload: p, qos, retain, - }) + })) .await; } +// Enqueue a subscribe request +pub async fn mqtt_subscribe(topic: &str) { + let mut t: HString = HString::new(); + let _ = t.push_str(&topic[..core::cmp::min(topic.len(), TOPIC_MAX)]); + CMD_CHAN.send(MqttCommand::Subscribe(t)).await; +} + +// Receiver for incoming MQTT messages (use in main) +pub fn mqtt_events( +) -> Receiver<'static, CriticalSectionRawMutex, IncomingMsg, EVENT_QUEUE> { + EVT_CHAN.receiver() +} + fn build_client_config() -> ClientConfig<'static, 8, CountingRng> { let rng = CountingRng(0); let mut cfg: ClientConfig<'static, 8, _> = ClientConfig::new(MqttVersion::MQTTv5, rng); @@ -105,34 +140,86 @@ async fn connect_mqtt(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> { client.connect_to_broker().await } -// While connected, either send queued publishes or ping periodically +// While connected, handle publishes, subscribes, ping, and incoming messages async fn connected_loop(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> { + // Subscribe to a default topic on connect (optional). You can remove this + // if you always subscribe from main via mqtt_subscribe(). + 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); + // Not fatal: continue + } + } + loop { - match select(PUB_CHAN.receive(), Timer::after(PING_PERIOD)).await { - // Got a publish request from main - Either::First(cmd) => { - // QoS1 may read PUBACK, keep all socket I/O here - let res = client - .send_message( - cmd.topic.as_str(), - &cmd.payload, - cmd.qos, - cmd.retain, - ) - .await; - if let Err(e) = res { + // Nested select so we react to incoming publishes ASAP, while still + // sending keepalive pings on schedule and servicing commands. + let in_or_ping = async { + select(client.receive_message(), Timer::after(PING_PERIOD)).await + }; + + match select(CMD_CHAN.receive(), in_or_ping).await { + // Command from main (publish/subscribe) + Either::First(cmd) => match cmd { + MqttCommand::Publish(m) => { + if let Err(e) = client + .send_message( + m.topic.as_str(), + &m.payload, + m.qos, + m.retain, + ) + .await + { + return Err(e); + } + } + MqttCommand::Subscribe(topic) => { + match client.subscribe_to_topic(topic.as_str()).await { + Ok(_) => info!("Subscribed to '{}'", topic.as_str()), + Err(e) => { + info!( + "Subscribe failed for '{}': {:?}", + topic.as_str(), + e + ); + return Err(e); + } + } + } + }, + // Either an incoming publish or the ping timer fired + Either::Second(e) => match e { + // Got a PUBLISH from broker + Either::First(Ok((topic, msg))) => { + let mut t: HString = HString::new(); + let _ = t.push_str( + &topic[..core::cmp::min(topic.len(), TOPIC_MAX)], + ); + let mut p: HVec = HVec::new(); + let take = core::cmp::min(msg.len(), PAYLOAD_MAX); + let _ = p.extend_from_slice(&msg[..take]); + EVT_CHAN + .send(IncomingMsg { topic: t, payload: p }) + .await; + } + // Error receiving => reconnect + Either::First(Err(e)) => { + info!("MQTT receive error (reconnect): {:?}", e); return Err(e); } - } - // Time to ping - Either::Second(_) => { - client.send_ping().await?; - } + // Ping timer fired + Either::Second(_) => { + client.send_ping().await?; + } + }, } } } -// One full MQTT session: TCP connect -> MQTT connect -> (optional) publish -> ping loop +// Full session: TCP connect -> MQTT connect -> connected loop async fn run_one_session( stack: Stack<'static>, tcp_rx: &mut [u8], @@ -140,13 +227,11 @@ async fn run_one_session( mqtt_tx: &mut [u8], mqtt_rx: &mut [u8], ) -> Result<(), ()> { - // Build socket and connect TCP let mut socket = TcpSocket::new(stack, tcp_rx, tcp_tx); if connect_tcp(&mut socket).await.is_err() { return Err(()); } - // Build client and connect MQTT let mut client = build_client(socket, mqtt_tx, mqtt_rx); match connect_mqtt(&mut client).await { Ok(_) => info!("MQTT CONNACK received"), @@ -156,10 +241,7 @@ async fn run_one_session( } } - // While connected, process publishes and ping. Any error triggers reconnect. - connected_loop(&mut client) - .await - .map_err(|_| ()) + connected_loop(&mut client).await.map_err(|_| ()) } #[embassy_executor::task] @@ -167,10 +249,10 @@ pub async fn mqtt_task(stack: Stack<'static>) { info!("MQTT task starting..."); // Take static buffers once and reuse across reconnects - let mut tcp_rx = TCP_RX_BUFFER.take(); - let mut tcp_tx = TCP_TX_BUFFER.take(); - let mut mqtt_tx = MQTT_TX_BUF.take(); - let mut mqtt_rx = MQTT_RX_BUF.take(); + 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( diff --git a/flake.nix b/flake.nix index 20709d9..28c1717 100644 --- a/flake.nix +++ b/flake.nix @@ -23,6 +23,7 @@ rustup cargo-espflash espup + mosquitto ]; shellHook = '' diff --git a/test1/src/bin/mqtt_docs.md b/test1/src/bin/mqtt_docs.md deleted file mode 100644 index 70403d2..0000000 --- a/test1/src/bin/mqtt_docs.md +++ /dev/null @@ -1,13 +0,0 @@ -# MQTT Implementation Notes - -* WiFi connection runs in a separate task with auto-reconnect functionality -* Network stack operates in a dedicated packet processing task -* MQTT uses TCP socket connection (MqttClient wraps this socket) -* ClientConfig specifies: MQTT version, QoS levels, client ID -* Standard flow: connect_to_broker() → send_message() or subscribe() - -## Implementation: -1. Perform DNS lookup for broker hostname -2. Establish TCP socket connection -3. Create MqttClient with configuration -4. Implement connect, publish, and subscribe logic