[Rust] [Connector] Fix status updates in base connector (#801)

# Problem
only vectors on statuses are applied as patch merges on the service, all
other fields need to be specified as the current state to not be
overwritten

# This PR
Updates status reporting APIs to send existing status fields when only
trying to do a partial update.
Note that some of this logic may change in the future, depending on how
the merging strategy evolves

Author: vaavva

rust/azure_iot_operations_connector/src/base_connector/managed_azure_device_registry.rs

@@ -279,14 +279,14 @@ impl DeviceEndpointClient {
     /// Used to report the status of just the device,
     /// and then updates the [`Device`] with the new status returned
     pub async fn report_device_status(&mut self, device_status: Result<(), AdrConfigError>) {
-        // Create status with empty endpoint status
+        // Create status without updating the endpoint status
         let status = azure_device_registry::DeviceStatus {
             config: Some(azure_device_registry::StatusConfig {
                 version: self.specification.version,
                 error: device_status.err(),
                 last_transition_time: None, // this field will be removed, so we don't need to worry about it for now
             }),
-            // inserts the inbound endpoint name with None if there's no error, or Some(AdrConfigError) if there is
+            // Endpoints are merged on the service, so sending an empty map won't update anything
             endpoints: HashMap::new(),
         };
 
@@ -296,10 +296,23 @@ impl DeviceEndpointClient {
 
     /// Used to report the status of just the endpoint,
     /// and then updates the [`Device`] with the new status returned
+    /// # Panics
+    /// if the status mutex has been poisoned, which should not be possible
     pub async fn report_endpoint_status(&mut self, endpoint_status: Result<(), AdrConfigError>) {
-        // Create status with empty device status
+        // If the version of the current status config matches the current version, then include the existing config.
+        // If there's no current config or the version doesn't match, don't report a status since the status for this version hasn't been reported yet
+        let current_config = self.status.read().unwrap().as_ref().and_then(|status| {
+            if status.config.as_ref().and_then(|config| config.version)
+                == self.specification.version
+            {
+                status.config.clone()
+            } else {
+                None
+            }
+        });
+        // Create status without updating the device status
         let status = azure_device_registry::DeviceStatus {
-            config: None,
+            config: current_config,
             // inserts the inbound endpoint name with None if there's no error, or Some(AdrConfigError) if there is
             endpoints: HashMap::from([(
                 self.device_endpoint_ref.inbound_endpoint_name.clone(),
@@ -831,16 +844,32 @@ impl DatasetClient {
     }
 
     /// Used to report the status of a dataset
+    /// # Panics
+    /// if the asset status mutex has been poisoned, which should not be possible
     pub async fn report_status(&self, status: Result<(), AdrConfigError>) {
+        // If the version of the current status config matches the current version, then include the existing config.
+        // If there's no current config or the version doesn't match, don't report a status since the status for this version hasn't been reported yet
+        let current_asset_config = self
+            .asset_status
+            .read()
+            .unwrap()
+            .as_ref()
+            .and_then(|status| {
+                if status.config.as_ref().and_then(|config| config.version)
+                    == self.asset_specification.version
+                {
+                    status.config.clone()
+                } else {
+                    None
+                }
+            });
+        // Get current message schema reference, so that it isn't overwritten
+        let current_message_schema_reference = self.message_schema_reference();
         let adr_asset_status = azure_device_registry::AssetStatus {
-            // TODO: Do I need to include the version here?
-            // config: Some(azure_device_registry::StatusConfig {
-            //     version: self.asset_specification.version,
-            //     ..azure_device_registry::StatusConfig::default()
-            // }),
+            config: current_asset_config,
             datasets: Some(vec![azure_device_registry::DatasetEventStreamStatus {
                 name: self.dataset_ref.dataset_name.clone(),
-                message_schema_reference: None,
+                message_schema_reference: current_message_schema_reference,
                 error: status.err(),
             }]),
             ..azure_device_registry::AssetStatus::default()
@@ -868,6 +897,8 @@ impl DatasetClient {
     /// # Panics
     /// If the Schema Registry Service returns a schema without required values. This should get updated
     /// to be validated by the Schema Registry API surface in the future
+    ///
+    /// If the asset status mutex has been poisoned, which should not be possible
     pub async fn report_message_schema(
         &self,
         message_schema: MessageSchema,
@@ -918,17 +949,42 @@ impl DatasetClient {
                     .expect("schema namespace will always be present."), // waiting on change to service DTDL for this to be guaranteed in code
             }
         })?;
-
+        // If the version of the current status config matches the current version, then include the existing config.
+        // If there's no current config or the version doesn't match, don't report a status since the status for this version hasn't been reported yet
+        let current_asset_config = self
+            .asset_status
+            .read()
+            .unwrap()
+            .as_ref()
+            .and_then(|status| {
+                if status.config.as_ref().and_then(|config| config.version)
+                    == self.asset_specification.version
+                {
+                    status.config.clone()
+                } else {
+                    None
+                }
+            });
+        // Get the current dataset config error, if it exists, so that it isn't overwritten
+        let current_dataset_config_error =
+            self.asset_status
+                .read()
+                .unwrap()
+                .as_ref()
+                .and_then(|status| {
+                    status.datasets.as_ref().and_then(|datasets| {
+                        datasets
+                            .iter()
+                            .find(|dataset| dataset.name == self.dataset_ref.dataset_name)
+                            .and_then(|dataset| dataset.error.clone())
+                    })
+                });
         let adr_asset_status = azure_device_registry::AssetStatus {
-            // TODO: Do I need to include the version here?
-            // config: Some(azure_device_registry::StatusConfig {
-            //     version: self.asset_specification.version,
-            //     ..azure_device_registry::StatusConfig::default()
-            // }),
+            config: current_asset_config,
             datasets: Some(vec![azure_device_registry::DatasetEventStreamStatus {
                 name: self.dataset_ref.dataset_name.clone(),
                 message_schema_reference: Some(message_schema_reference.clone()),
-                error: None,
+                error: current_dataset_config_error,
             }]),
             ..azure_device_registry::AssetStatus::default()
         };