wait for conn ack to start listener loop

tester/src/main.rs

@@ -21,11 +21,26 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .about("CLN:mqtt-tester - MQTT client signer")
         .arg(Arg::from("--test run a test against the embedded device"));
 
-    let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
+    let mut try_i = 0;
-    mqttoptions.set_credentials(USERNAME, PASSWORD);
+    let (client, mut eventloop) = loop {
-    mqttoptions.set_keep_alive(Duration::from_secs(5));
+        let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
-
+        mqttoptions.set_credentials(USERNAME, PASSWORD);
-    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+        mqttoptions.set_keep_alive(Duration::from_secs(5));
+        let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+        match eventloop.poll().await {
+            Ok(event) => {
+                if let Some(_) = incoming_conn_ack(event) {
+                    println!("==========> MQTT connected!");
+                    break (client, eventloop);
+                }
+            }
+            Err(_) => {
+                try_i = try_i + 1;
+                println!("reconnect.... {}", try_i);
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        }
+    };
 
     client
         .subscribe(SUB_TOPIC, QoS::AtMostOnce)
@@ -109,3 +124,12 @@ fn incoming_bytes(event: Event) -> Option<Vec<u8>> {
     }
     None
 }
+
+fn incoming_conn_ack(event: Event) -> Option<()> {
+    if let Event::Incoming(packet) = event {
+        if let Packet::ConnAck(_) = packet {
+            return Some(());
+        }
+    }
+    None
+}