disconnect from local mqtt broker cleanly

Signed-off-by: Reuben Miller <[email protected]>

crates/core/tedge/src/cli/connect/command.rs

@@ -289,6 +289,7 @@ fn check_device_status_c8y(tedge_config: &TEdgeConfig) -> Result<DeviceStatus, C
         .network_options
         .set_connection_timeout(CONNECTION_TIMEOUT.as_secs());
     let mut acknowledged = false;
+    let mut exists = false;
 
     client.subscribe(&c8y_topic_builtin_jwt_token_downstream, AtLeastOnce)?;
 
@@ -312,7 +313,8 @@ fn check_device_status_c8y(tedge_config: &TEdgeConfig) -> Result<DeviceStatus, C
                 let token = String::from_utf8(response.payload.to_vec()).unwrap();
                 // FIXME: what does this magic number mean?
                 if token.contains("71") {
-                    return Ok(DeviceStatus::AlreadyExists);
+                    exists = true;
+                    break;
                 }
             }
             Ok(Event::Outgoing(Outgoing::PingReq)) => {
@@ -332,6 +334,19 @@ fn check_device_status_c8y(tedge_config: &TEdgeConfig) -> Result<DeviceStatus, C
         }
     }
 
+    // Cleanly disconnect client
+    client.disconnect()?;
+    for event in connection.iter() {
+        match event {
+            Ok(Event::Outgoing(Outgoing::Disconnect)) | Err(_) => break,
+            _ => {}
+        }
+    }
+
+    if exists {
+        return Ok(DeviceStatus::AlreadyExists);
+    }
+
     if acknowledged {
         // The request has been sent but without a response
         Ok(DeviceStatus::Unknown)
@@ -363,6 +378,7 @@ fn check_device_status_azure(tedge_config: &TEdgeConfig) -> Result<DeviceStatus,
 
     let (mut client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
     let mut acknowledged = false;
+    let mut exists = false;
 
     client.subscribe(AZURE_TOPIC_DEVICE_TWIN_DOWNSTREAM, AtLeastOnce)?;
 
@@ -385,7 +401,8 @@ fn check_device_status_azure(tedge_config: &TEdgeConfig) -> Result<DeviceStatus,
                 // We got a response
                 if response.topic.contains(REGISTRATION_OK) {
                     println!("Received expected response message.");
-                    return Ok(DeviceStatus::AlreadyExists);
+                    exists = true;
+                    break;
                 } else {
                     break;
                 }
@@ -407,6 +424,19 @@ fn check_device_status_azure(tedge_config: &TEdgeConfig) -> Result<DeviceStatus,
         }
     }
 
+    // Cleanly disconnect client
+    client.disconnect()?;
+    for event in connection.iter() {
+        match event {
+            Ok(Event::Outgoing(Outgoing::Disconnect)) | Err(_) => break,
+            _ => {}
+        }
+    }
+
+    if exists {
+        return Ok(DeviceStatus::AlreadyExists);
+    }
+
     if acknowledged {
         // The request has been sent but without a response
         Ok(DeviceStatus::Unknown)
@@ -430,6 +460,7 @@ fn check_device_status_aws(tedge_config: &TEdgeConfig) -> Result<DeviceStatus, C
 
     let (mut client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
     let mut acknowledged = false;
+    let mut exists = false;
 
     client.subscribe(AWS_TOPIC_SUB_CHECK_CONNECTION, AtLeastOnce)?;
 
@@ -451,7 +482,8 @@ fn check_device_status_aws(tedge_config: &TEdgeConfig) -> Result<DeviceStatus, C
             Ok(Event::Incoming(Packet::Publish(response))) => {
                 // We got a response
                 println!("Received expected response on topic {}.", response.topic);
-                return Ok(DeviceStatus::AlreadyExists);
+                exists = true;
+                break;
             }
             Ok(Event::Outgoing(Outgoing::PingReq)) => {
                 // No messages have been received for a while
@@ -470,6 +502,19 @@ fn check_device_status_aws(tedge_config: &TEdgeConfig) -> Result<DeviceStatus, C
         }
     }
 
+    // Cleanly disconnect client
+    client.disconnect()?;
+    for event in connection.iter() {
+        match event {
+            Ok(Event::Outgoing(Outgoing::Disconnect)) | Err(_) => break,
+            _ => {}
+        }
+    }
+
+    if exists {
+        return Ok(DeviceStatus::AlreadyExists);
+    }
+
     if acknowledged {
         // The request has been sent but without a response
         Ok(DeviceStatus::Unknown)

crates/core/tedge/src/cli/connect/jwt_token.rs

@@ -27,6 +27,7 @@ pub(crate) fn get_connected_c8y_url(tedge_config: &TEdgeConfig) -> Result<String
         .network_options
         .set_connection_timeout(CONNECTION_TIMEOUT.as_secs());
     let mut acknowledged = false;
+    let mut c8y_url: Option<String> = None;
 
     client.subscribe(&c8y_topic_builtin_jwt_token_downstream, AtLeastOnce)?;
 
@@ -49,7 +50,8 @@ pub(crate) fn get_connected_c8y_url(tedge_config: &TEdgeConfig) -> Result<String
                 // We got a response
                 let token = String::from_utf8(response.payload.to_vec()).unwrap();
                 let connected_url = decode_jwt_token(token.as_str())?;
-                return Ok(connected_url);
+                c8y_url = Some(connected_url);
+                break;
             }
             Ok(Event::Outgoing(Outgoing::PingReq)) => {
                 // No messages have been received for a while
@@ -68,6 +70,19 @@ pub(crate) fn get_connected_c8y_url(tedge_config: &TEdgeConfig) -> Result<String
         }
     }
 
+    // Cleanly disconnect client
+    client.disconnect()?;
+    for event in connection.iter() {
+        match event {
+            Ok(Event::Outgoing(Outgoing::Disconnect)) | Err(_) => break,
+            _ => {}
+        }
+    }
+
+    if let Some(c8y_url) = c8y_url {
+        return Ok(c8y_url);
+    }
+
     if acknowledged {
         // The request has been sent but without a response
         println!("\nThe request has been sent, however, no response.");