Stop sending empty StreamProperties for each new steam (#34)

quixio · Feb 13, 2024 · a689b55 · a689b55
1 parent c10e1ef
commit a689b55
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 6 deletions.
diff --git a/src/QuixStreams.Streaming/Models/StreamProducer/StreamPropertiesProducer.cs b/src/QuixStreams.Streaming/Models/StreamProducer/StreamPropertiesProducer.cs
@@ -23,7 +23,8 @@ public class StreamPropertiesProducer : IStreamPropertiesProducer
         private readonly Timer flushTimer;
         private bool timerEnabled = false; // Here because every now and then resetting its due time to never doesn't work
         private const int PropertyChangedFlushInterval = 20;
-        private int lastHash = 0;
+        private const int DefaultEmptyHash = 0;
+        private int lastHash = DefaultEmptyHash;
         private readonly object flushLock = new object();
         private bool isDisposed = false;
 
@@ -204,10 +205,12 @@ private void Flush(bool force, bool flushOnlyOnChange = true)
                 };
 
                 var hash = streamProperties.GetHashCode();
-                if (flushOnlyOnChange && hash == lastHash)
-                {
-                    return;
-                }
+
+                // Has it changed
+                if (flushOnlyOnChange && hash == lastHash) return;
+
+                // Do not flush empty state for a new stream
+                if (lastHash == DefaultEmptyHash && !streamProperties.IsSet()) return;
 
                 this.lastHash = hash;
                 this.lastHeartbeatRebroadcastTime = DateTimeOffset.Now.ToUnixTimeMilliseconds();
@@ -225,7 +228,7 @@ List<string> GetParents()
                 {
                     try
                     {
-                        return this.Parents.ToList();
+                        return this.Parents.Distinct().ToList();
                     }
                     catch (System.ArgumentException ex)
                     {

diff --git a/src/QuixStreams.Telemetry/Models/Telemetry/StreamProperties/StreamProperties.cs b/src/QuixStreams.Telemetry/Models/Telemetry/StreamProperties/StreamProperties.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace QuixStreams.Telemetry.Models
 {
@@ -35,6 +36,20 @@ public class StreamProperties
         /// </summary>
         public DateTime? TimeOfRecording { get; set; }
 
+        /// <summary>
+        /// Returns whether any values differ from empty value
+        /// </summary>
+        /// <returns></returns>
+        public bool IsSet()
+        {
+            if (!string.IsNullOrWhiteSpace(this.Name)) return true;
+            if (!string.IsNullOrWhiteSpace(this.Location)) return true;
+            if (this.Metadata != null && this.Metadata.Count > 0) return true;
+            if (this.Parents != null && this.Parents.Count > 0) return true;
+            if (this.TimeOfRecording != null) return true;
+            return false;
+        }
+
         /// <summary>
         /// Returns the hash of the content
         /// </summary>