Collector/connector reactive pipeline minor refactoring (#271)

home-climate-control · Sep 23, 2023 · 0f95ce5 · 0f95ce5
1 parent 6d5b731
commit 0f95ce5
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/dz3r-director/src/main/java/net/sf/dz3r/model/UnitDirector.java b/dz3r-director/src/main/java/net/sf/dz3r/model/UnitDirector.java
@@ -91,16 +91,19 @@ public UnitDirector(
 
         Optional.ofNullable(metricsCollectorSet)
                 .ifPresent(collectors -> Flux.fromIterable(collectors)
+                        .publishOn(Schedulers.boundedElastic())
                         .doOnNext(c -> c.connect(feed))
-                        .subscribeOn(Schedulers.boundedElastic())
+                        .doOnComplete(() -> logger.info("{}: connected metric collectors", getAddress()))
                         .subscribe());
+
         Optional.ofNullable(connectorSet)
                 .ifPresent(connectors -> Flux.fromIterable(connectors)
+                        .publishOn(Schedulers.boundedElastic())
                         .doOnNext(c -> {
                             c.connect(feed);
                             // VT: FIXME: Connect the control input when the API signature is established
                         })
-                        .subscribeOn(Schedulers.boundedElastic())
+                        .doOnComplete(() -> logger.info("{}: connected connectors", getAddress()))
                         .subscribe());
 
         logger.info("Configured: {} ({} zones: {})",

diff --git a/dz3r-http-gae/src/main/java/net/sf/dz3r/view/http/gae/v3/HttpConnectorGAE.java b/dz3r-http-gae/src/main/java/net/sf/dz3r/view/http/gae/v3/HttpConnectorGAE.java
@@ -97,7 +97,6 @@ public void connect(UnitDirector.Feed feed) {
                 );
 
         zoneStatusSubscription = zoneStatusFeed
-                .publishOn(Schedulers.boundedElastic())
                 .buffer(pollInterval)
                 .doOnNext(this::exchange)
                 .subscribe();