Bugfix: observing the schedule without the console as well (#271, #281)

dz3r-bootstrap/src/main/java/net/sf/dz3r/runtime/config/ConfigurationParser.java

@@ -1,5 +1,7 @@
 package net.sf.dz3r.runtime.config;
 
+import net.sf.dz3r.instrumentation.InstrumentCluster;
+import net.sf.dz3r.instrumentation.Marker;
 import net.sf.dz3r.runtime.config.connector.ConnectorConfigurationParser;
 import net.sf.dz3r.runtime.config.filter.FilterConfigurationParser;
 import net.sf.dz3r.runtime.config.hardware.HvacConfigurationParser;
@@ -11,8 +13,6 @@
 import net.sf.dz3r.runtime.config.mqtt.MqttConfigurationParser;
 import net.sf.dz3r.runtime.config.onewire.OnewireConfigurationParser;
 import net.sf.dz3r.runtime.config.schedule.ScheduleConfigurationParser;
-import net.sf.dz3r.instrumentation.InstrumentCluster;
-import net.sf.dz3r.instrumentation.Marker;
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -109,6 +109,7 @@ public ConfigurationContext parse(HccRawConfig source) {
             var ic = new InstrumentCluster(
                     ctx.sensors.getFlux(),
                     ctx.switches.getFlux(),
+                    ctx.schedule.getFlux(),
                     ctx.connectors.getFlux(),
                     ctx.collectors.getFlux(),
                     ctx.hvacDevices.getFlux());

dz3r-bootstrap/src/main/java/net/sf/dz3r/runtime/config/model/DirectorConfigurationParser.java

@@ -1,8 +1,8 @@
 package net.sf.dz3r.runtime.config.model;
 
+import net.sf.dz3r.model.UnitDirector;
 import net.sf.dz3r.runtime.config.ConfigurationContext;
 import net.sf.dz3r.runtime.config.ConfigurationContextAware;
-import net.sf.dz3r.model.UnitDirector;
 import net.sf.dz3r.scheduler.ScheduleUpdater;
 import net.sf.dz3r.view.Connector;
 import net.sf.dz3r.view.MetricsCollector;

dz3r-director/build.gradle.kts

@@ -1,5 +1,6 @@
 dependencies {
 
+    api(project(":dz3r-common"))
     api(project(":dz3r-model"))
     api(project(":dz3r-scheduler"))
 }

dz3r-director/src/main/java/net/sf/dz3r/instrumentation/InstrumentCluster.java

@@ -2,6 +2,7 @@
 
 import net.sf.dz3r.device.actuator.HvacDevice;
 import net.sf.dz3r.device.actuator.Switch;
+import net.sf.dz3r.scheduler.ScheduleUpdater;
 import net.sf.dz3r.signal.Signal;
 import net.sf.dz3r.signal.health.SystemStatus;
 import net.sf.dz3r.view.Connector;
@@ -30,6 +31,7 @@ public class InstrumentCluster {
 
     private final Flux<Map.Entry<String, Flux<Signal<Double, Void>>>> sensors;
     private final Flux<Map.Entry<String, Switch<?>>> switches;
+    private final Flux<Map.Entry<String, ScheduleUpdater>> schedule;
     private final Flux<Map.Entry<String, Connector>> connectors;
     private final Flux<Map.Entry<String, MetricsCollector>> collectors;
     private final Flux<Map.Entry<String, HvacDevice>> hvacDevices;
@@ -49,13 +51,15 @@ public class InstrumentCluster {
     public InstrumentCluster(
             Flux<Map.Entry<String, Flux<Signal<Double, Void>>>> sensors,
             Flux<Map.Entry<String, Switch<?>>> switches,
+            Flux<Map.Entry<String, ScheduleUpdater>> schedule,
             Flux<Map.Entry<String, Connector>> connectors,
             Flux<Map.Entry<String, MetricsCollector>> collectors,
             Flux<Map.Entry<String, HvacDevice>> hvacDevices
             ) {
 
         this.sensors = sensors;
         this.switches = switches;
+        this.schedule = schedule;
         this.connectors = connectors;
         this.collectors = collectors;
         this.hvacDevices = hvacDevices;
@@ -71,6 +75,7 @@ public Flux<Signal<SystemStatus, Void>> getFlux() {
         connectSwitches();
 
         logger.error("FIXME: NOT IMPLEMENTED: getFlux(dampers)");
+        logger.error("FIXME: NOT IMPLEMENTED: getFlux(schedule)");
         logger.error("FIXME: NOT IMPLEMENTED: getFlux(collectors)");
         logger.error("FIXME: NOT IMPLEMENTED: getFlux(connectors)");
 

dz3r-director/src/main/java/net/sf/dz3r/model/UnitDirector.java

@@ -30,7 +30,7 @@
 /**
  * Assembles all the components related to one hardware HVAC unit, connects them, and manages their lifecycles.
  *
- * @author Copyright &copy; <a href="mailto:[email protected]">Vadim Tkachenko</a> 2001-2021
+ * @author Copyright &copy; <a href="mailto:[email protected]">Vadim Tkachenko</a> 2001-2023
  */
 public class UnitDirector implements Addressable<String>, AutoCloseable {
 
@@ -70,7 +70,20 @@ public UnitDirector(
 
         var scheduleFlux = Optional.ofNullable(scheduleUpdater)
                 .map(u -> connectScheduler(sensorFlux2zone.values(), u))
-                .orElse(Flux.empty());
+                .orElseGet(() -> {
+                    logger.warn("{}: no scheduler provided, running defaults", getAddress());
+                    return Flux.empty();
+                });
+
+        // This is necessary because... <facepalm> the architecture is screwed up and applying the schedule to the zone
+        // is a side effect of consuming this flux. No wonder the schedule was only applied when the console was up,
+        // it was the only consumer until now.
+        // See https://github.com/home-climate-control/dz/issues/281
+
+        scheduleFlux
+                .publishOn(Schedulers.newSingle("schedule-watcher-" + name))
+                .doOnNext(s -> logger.debug("{}: zone={}, event={}", name, s.getKey(), s.getValue()))
+                .subscribe();
 
         feed = connectFeeds(sensorFlux2zone, unitController, hvacDevice, hvacMode, scheduleFlux);
 
@@ -146,16 +159,15 @@ private Feed connectFeeds(
 
     private Flux<Map.Entry<String, Map.Entry<SchedulePeriod, ZoneSettings>>> connectScheduler(Collection<Zone> zones, ScheduleUpdater scheduleUpdater) {
 
-        if (scheduleUpdater == null) {
-            logger.warn("no scheduler provided, running defaults");
-            return Flux.empty();
-        }
+        net.sf.dz3r.common.HCCObjects.requireNonNull(scheduleUpdater, "programming error, this should've been resolved up the call stack");
 
         var name2zone = Flux.fromIterable(zones)
                 .map(z -> new AbstractMap.SimpleEntry<>(z.getAddress(), z))
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
                 .block();
 
+        logger.info("{}: connected schedules: {}", getAddress(), name2zone);
+
         var scheduler = new Scheduler(name2zone);
 
         return scheduler.connect(scheduleUpdater.update());

dz3r-scheduler/src/main/java/net/sf/dz3r/scheduler/Scheduler.java

@@ -22,7 +22,7 @@
  * Listens to the schedule event feed coming from {@link ScheduleUpdater}, and
  * passes the commands down to {@link net.sf.dz3r.model.Zone}s.
  *
- * @author Copyright &copy; <a href="mailto:[email protected]">Vadim Tkachenko</a> 2001-2021
+ * @author Copyright &copy; <a href="mailto:[email protected]">Vadim Tkachenko</a> 2001-2023
  */
 public class Scheduler {
 
@@ -59,17 +59,17 @@ public Flux<Map.Entry<String, Map.Entry<SchedulePeriod, ZoneSettings>>> connect(
 
         // Observe
         var observe = source
+                .publishOn(Schedulers.newSingle("schedule-observe"))
                 .flatMap(this::updateSchedule)
                 .flatMap(this::applySchedule)
-                .doOnNext(s -> logger.info("scheduleObserve: {}", s))
-                .subscribeOn(Schedulers.boundedElastic());
+                .doOnNext(s -> logger.info("scheduleObserve: {}", s));
 
         // Execute
         var execute = Flux.interval(scheduleGranularity, Schedulers.boundedElastic())
+                .publishOn(Schedulers.newSingle("schedule-execute"))
                 .flatMap(s -> Flux.fromIterable(zone2schedule.entrySet()))
                 .flatMap(this::applySchedule)
-                .doOnNext(s -> logger.info("scheduleExecute: {}", s))
-                .subscribeOn(Schedulers.boundedElastic());
+                .doOnNext(s -> logger.info("scheduleExecute: {}", s));
 
         return Flux
                 .merge(observe, execute)
@@ -91,6 +91,22 @@ private Flux<Map.Entry<Zone, SortedMap<SchedulePeriod, ZoneSettings>>> updateSch
         return Flux.just(new AbstractMap.SimpleEntry<>(zone, schedule));
     }
 
+    /**
+     * Apply a period to a zone, possibly.
+     *
+     * @param source Mapping from a zone to a set of their periods.
+     *
+     * @return Status update. Values are as follows:
+     *
+     * <ul>
+     *     <li>Nothing if the zone is on hold;</li>
+     *     <li>Nothing if the zone is already at current period;</li>
+     *     <li>Mapping of the zone name to period and settings upon a change;</li>
+     *     <li>Mapping of the zone name to {@code null} if there is no active period.</li>
+     * </ul>
+     *
+     * VT: FIXME: The above looks weird, need refactoring.
+     */
     private synchronized Flux<Map.Entry<String, Map.Entry<SchedulePeriod, ZoneSettings>>> applySchedule(Map.Entry<Zone, SortedMap<SchedulePeriod, ZoneSettings>> source) {
 
         ThreadContext.push("applySchedule");