End ComponentData streams that have no receivers

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>

Author: shsms

src/client/microgrid_client_actor.rs

@@ -27,6 +27,7 @@ use crate::{
 enum StreamStatus {
     Failed(u64),
     Connected(u64),
+    Ended(u64),
 }
 
 /// This actor owns the connection to the microgrid API and processes instructions
@@ -85,6 +86,9 @@ impl MicrogridClientActor {
                         Some(StreamStatus::Connected(component_id)) => {
                             components_to_retry.remove(&component_id);
                         }
+                        Some(StreamStatus::Ended(component_id)) => {
+                            components_to_retry.remove(&component_id);
+                        }
                         None => {
                             tracing::error!("MicrogridClientActor: Stream status channel closed, exiting.");
                             return;
@@ -263,6 +267,23 @@ async fn run_component_data_stream(
     stream_stopped_tx: mpsc::Sender<StreamStatus>,
 ) {
     loop {
+        if tx.receiver_count() == 0 {
+            tracing::debug!(
+                "Dropping ComponentData stream for component_id:{:?}",
+                component_id
+            );
+            stream_stopped_tx
+                .send(StreamStatus::Ended(component_id))
+                .await
+                .unwrap_or_else(|e| {
+                    tracing::error!(
+                        "Failed to send stream ended message for {:?}: {:?}",
+                        component_id,
+                        e
+                    );
+                });
+            return;
+        }
         let message = match stream.message().await {
             Ok(m) => m,
             Err(e) => {
@@ -289,16 +310,8 @@ async fn run_component_data_stream(
             }
         };
 
-        match tx.send(data) {
-            Ok(_) => {}
-            Err(e) => {
-                tracing::error!(
-                    "Unable to send component data for {:?}: {}.  Closing stream.",
-                    component_id,
-                    e
-                );
-                break;
-            }
+        if tx.send(data).is_err() {
+            continue;
         };
     }