Sending incremental update instead of complete (#268)

home-climate-control · Aug 4, 2023 · b12a941 · b12a941
1 parent bce92d8
commit b12a941
Showing 1 changed file with 19 additions and 9 deletions.
diff --git a/dz3r-director/src/main/java/net/sf/dz3r/instrumentation/InstrumentCluster.java b/dz3r-director/src/main/java/net/sf/dz3r/instrumentation/InstrumentCluster.java
@@ -40,12 +40,7 @@ public class InstrumentCluster {
      *
      * This object gets updated and then emitted every time an update comes.
      */
-    private final SystemStatus currentStatus = new SystemStatus(
-            new TreeMap<>(),
-            new TreeMap<>(),
-            new TreeMap<>(),
-            new TreeMap<>(),
-            new TreeMap<>());
+    private final SystemStatus currentStatus = createEmptyStatus();
 
     private final Sinks.Many<Signal<SystemStatus, Void>> statusSink = Sinks.many().multicast().onBackpressureBuffer();
 
@@ -79,16 +74,22 @@ public Flux<Signal<SystemStatus, Void>> getFlux() {
                 })
                 .subscribe(kv -> {
 
-                    var id = kv.getKey();
+                    String id = kv.getKey();
                     var status = kv.getValue();
 
                     status
                             .subscribe(s -> {
 
                                 logger.info("update: id={}, status={}", id, s);
-                                currentStatus.sensors().put(String.valueOf(id), s.getValue());
 
-                                statusSink.tryEmitNext(new Signal<>(Instant.now(), currentStatus));
+                                // Update the accumulated status
+                                currentStatus.sensors().put(id, s.getValue());
+
+                                // Send an incremental update
+                                var incrementalStatus = createEmptyStatus();
+                                incrementalStatus.sensors().put(id, s.getValue());
+
+                                statusSink.tryEmitNext(new Signal<>(Instant.now(), incrementalStatus));
                             });
 
                 });
@@ -97,4 +98,13 @@ public Flux<Signal<SystemStatus, Void>> getFlux() {
 
         return statusSink.asFlux();
     }
+
+    private SystemStatus createEmptyStatus() {
+        return new SystemStatus(
+                new TreeMap<>(),
+                new TreeMap<>(),
+                new TreeMap<>(),
+                new TreeMap<>(),
+                new TreeMap<>());
+    }
 }