reconnect loop on dropped connection

final/src/mqtt/client.rs

@@ -11,17 +11,21 @@ use static_cell::ConstStaticCell;
 
 use crate::mqtt::config::mqtt_broker_endpoint;
 
+const RECONNECT_DELAY_SECS: u64 = 5;
+
 // 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)
+// 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]);
 
 // Type alias for clarity in helper functions
-// TODO number 8 is number of allowed subs
+// NOTE: Tie the TcpSocket lifetime to the session (not 'static), and
-type Client<'a> = MqttClient<'a, TcpSocket<'static>, 8, CountingRng>;
+// the MQTT buffers lifetime to 'a. This resolves the E0521 error.
+// The const 8 is the MAX_PROPERTIES generic used by rust-mqtt config.
+type Client<'a, 'net> = MqttClient<'a, TcpSocket<'net>, 8, CountingRng>;
 
 fn build_client_config() -> ClientConfig<'static, 8, CountingRng> {
     let rng = CountingRng(0);
@@ -31,18 +35,18 @@ fn build_client_config() -> ClientConfig<'static, 8, CountingRng> {
     cfg
 }
 
-fn build_client<'a>(
+fn build_client<'a, 'net>(
-    socket: TcpSocket<'static>,
+    socket: TcpSocket<'net>,
     mqtt_tx: &'a mut [u8],
     mqtt_rx: &'a mut [u8],
-) -> Client<'a> {
+) -> Client<'a, 'net> {
     let cfg = build_client_config();
     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, cfg)
 }
 
-async fn connect_tcp(socket: &mut TcpSocket<'static>) -> Result<(), ()> {
+async fn connect_tcp<'net>(socket: &mut TcpSocket<'net>) -> Result<(), ()> {
     match socket.connect(mqtt_broker_endpoint()).await {
         Ok(_) => {
             info!("Connected TCP to MQTT broker");
@@ -55,11 +59,11 @@ async fn connect_tcp(socket: &mut TcpSocket<'static>) -> Result<(), ()> {
     }
 }
 
-async fn connect_mqtt(client: &mut Client<'_>) -> Result<(), ReasonCode> {
+async fn connect_mqtt(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> {
     client.connect_to_broker().await
 }
 
-async fn publish_once(client: &mut Client<'_>) -> Result<(), ReasonCode> {
+async fn publish_once(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> {
     client
         .send_message(
             "esp32/topic",
@@ -70,7 +74,7 @@ async fn publish_once(client: &mut Client<'_>) -> Result<(), ReasonCode> {
         .await
 }
 
-async fn ping_loop(client: &mut Client<'_>) -> Result<(), ReasonCode> {
+async fn ping_loop(client: &mut Client<'_, '_>) -> Result<(), ReasonCode> {
     loop {
         if let Err(reason) = client.send_ping().await {
             info!("MQTT ping failed: {:?}", reason);
@@ -80,20 +84,18 @@ async fn ping_loop(client: &mut Client<'_>) -> Result<(), ReasonCode> {
     }
 }
 
-#[embassy_executor::task]
+// One full MQTT session: TCP connect -> MQTT connect -> (optional) publish -> ping loop
-pub async fn mqtt_task(stack: Stack<'static>) {
+async fn run_one_session(
-    info!("MQTT task starting...");
+    stack: Stack<'static>,
-
+    tcp_rx: &mut [u8],
-    // Take static buffers once
+    tcp_tx: &mut [u8],
-    let tcp_rx = TCP_RX_BUFFER.take();
+    mqtt_tx: &mut [u8],
-    let tcp_tx = TCP_TX_BUFFER.take();
+    mqtt_rx: &mut [u8],
-    let mqtt_tx = MQTT_TX_BUF.take();
+) -> Result<(), ()> {
-    let mqtt_rx = MQTT_RX_BUF.take();
-
     // Build socket and connect TCP
     let mut socket = TcpSocket::new(stack, tcp_rx, tcp_tx);
     if connect_tcp(&mut socket).await.is_err() {
-        return;
+        return Err(());
     }
 
     // Build client and connect MQTT
@@ -102,16 +104,44 @@ pub async fn mqtt_task(stack: Stack<'static>) {
         Ok(_) => info!("MQTT CONNACK received"),
         Err(reason) => {
             info!("MQTT connect failed: {:?}", reason);
-            return;
+            return Err(());
         }
     }
 
-    // Publish single message
+    // Optional demo publish (same behavior as before)
     match publish_once(&mut client).await {
         Ok(_) => info!("MQTT PUBLISH sent"),
         Err(reason) => info!("MQTT publish failed: {:?}", reason),
     }
 
-    // Keep the connection with pings
+    // Keepalive. Any error ends the session; outer loop will reconnect.
-    let _ = ping_loop(&mut client).await;
+    ping_loop(&mut client).await.map_err(|_| ())
+}
+
+#[embassy_executor::task]
+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();
+
+    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;
+    }
 }