fix: clear stream_metrics on write to network (#91)

* fix: don't send when `stream_metrics` is empty * fix: clear metrics only on successful publish
bytebeamio · Dec 30, 2022 · ee37777 · ee37777
1 parent 8bab9bd
commit ee37777
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 6 deletions.
diff --git a/uplink/src/base/serializer.rs b/uplink/src/base/serializer.rs
@@ -471,21 +471,28 @@ impl<C: MqttClient> Serializer<C> {
                 }
                 _ = interval.tick(), if metrics_enabled => {
                     if let Some(handler) = self.serializer_metrics.as_mut() {
-                        info!("Publishing serializer metrics to broker");
                         let data = handler.update();
                         let payload = serde_json::to_vec(&vec![data])?;
                         handler.clear();
-                        if let Err(e) = self.client.try_publish(&handler.topic, QoS::AtLeastOnce, false, payload) {
-                            error!("Couldn't publish serializer metrics to broker: {}", e)
+
+                        info!("Publishing serializer metrics to broker");
+                        match self.client.try_publish(&handler.topic, QoS::AtLeastOnce, false, payload) {
+                            Err(e) => error!("Couldn't publish serializer metrics to broker: {e}"),
+                            _ => handler.clear(),
                         }
                     }
 
                     if let Some(handler) = self.stream_metrics.as_mut() {
-                        info!("Publishing stream metrics to broker");
                         let data: Vec<&mut StreamMetrics> = handler.streams().collect();
+                        if data.is_empty() {
+                            continue;
+                        }
                         let payload = serde_json::to_vec(&data)?;
-                        if let Err(e) = self.client.try_publish(&handler.topic, QoS::AtLeastOnce, false, payload) {
-                            error!("Couldn't publish stream metrics to broker: {}", e)
+
+                        info!("Publishing stream metrics to broker");
+                        match self.client.try_publish(&handler.topic, QoS::AtLeastOnce, false, payload) {
+                            Err(e) => error!("Couldn't publish stream metrics to broker: {e}"),
+                            _ => handler.clear(),
                         }
                     }
                 }

diff --git a/uplink/src/base/serializer/metrics.rs b/uplink/src/base/serializer/metrics.rs
@@ -149,6 +149,10 @@ impl StreamMetricsHandler {
     pub fn streams(&mut self) -> Streams {
         Streams { values: self.map.values_mut() }
     }
+
+    pub fn clear(&mut self) {
+        self.map.clear();
+    }
 }
 
 pub struct Streams<'a> {