filipriec/projekt1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working send/receive to the broker

Browse Source
This commit is contained in:
Priec
2025-10-05 21:19:37 +02:00
parent 73cbf7f912
commit b4e86bded6
4 changed files with 168 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
final/src/bin/main.rs
View File

@@ -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);
}
}

162
final/src/mqtt/client.rs
View File

@@ -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<TOPIC_MAX>,
@@ -44,10 +47,28 @@ struct PublishMsg {
retain: bool,
}
static PUB_CHAN: Channel<CriticalSectionRawMutex, PublishMsg, PUBLISH_QUEUE> =
#[derive(Clone)]
enum MqttCommand {
Publish(PublishMsg),
Subscribe(HString<TOPIC_MAX>),
}
static CMD_CHAN: Channel<CriticalSectionRawMutex, MqttCommand, COMMAND_QUEUE> =
Channel::new();
// Public API for main: enqueue a publish
// ===== Events (messages) from MQTT task to main =====
#[derive(Clone)]
pub struct IncomingMsg {
pub topic: HString<TOPIC_MAX>,
pub payload: HVec<u8, PAYLOAD_MAX>,
}
static EVT_CHAN: Channel<CriticalSectionRawMutex, IncomingMsg, EVENT_QUEUE> =
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<u8, PAYLOAD_MAX> = 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<TOPIC_MAX> = 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<TOPIC_MAX> = HString::new();
let _ = t.push_str(
&topic[..core::cmp::min(topic.len(), TOPIC_MAX)],
);
let mut p: HVec<u8, PAYLOAD_MAX> = 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(