pollinator: Reload configuration after notify

Author: DougLau

Cargo.lock

@@ -15,8 +15,8 @@ use pollinator::CommLinkCfg;
 use resin::{Database, Error, Result};
 use std::collections::HashMap;
 use tokio::task::{AbortHandle, JoinSet};
-use tokio::time::{Duration, interval};
-use tokio_stream::StreamExt;
+use tokio::time::Duration;
+use tokio_postgres::Notification;
 
 /// Command-line arguments
 #[derive(FromArgs)]
@@ -84,34 +84,71 @@ async fn poll_comm_links(db: Database) -> Result<()> {
                 Some(cfg) => {
                     let changed = *cfg != task.cfg;
                     if changed {
+                        log::info!("{name}: changed, aborting");
                         // comm link changed: abort task
                         task.handle.abort();
                     }
                     !changed
                 }
                 None => {
+                    log::info!("{name}: removed, aborting");
                     // no comm link: abort task
                     task.handle.abort();
                     false
                 }
             }
         });
         // spawn tasks for new comm links
-        for cfg in cfgs {
+        for cfg in &cfgs {
             if !tasks.contains_key(cfg.name()) {
                 let name = cfg.name().to_string();
                 let db = Some(db.clone());
+                log::info!("{name}: spawning");
                 let handle = set.spawn(cfg.clone().run(db));
-                let task = CommLinkTask { cfg, handle };
+                let task = CommLinkTask {
+                    cfg: cfg.clone(),
+                    handle,
+                };
                 tasks.insert(name, task);
             }
         }
-        let _not = stream.next().await;
-        // FIXME: check notification channel / payload
+        while let Some(not) = stream.recv().await {
+            if should_reload(&not, &cfgs) {
+                log::info!(
+                    "{} {}: reloading configuration",
+                    not.channel(),
+                    not.payload()
+                );
+                tokio::time::sleep(Duration::from_secs(2)).await;
+                // Empty the notification stream
+                while !stream.is_empty() {
+                    stream.recv().await;
+                }
+                break;
+            }
+        }
     }
     Ok(())
 }
 
+/// Check if a notification should trigger reloading the configuration
+fn should_reload(not: &Notification, cfgs: &[CommLinkCfg]) -> bool {
+    // FIXME: comm_link "connected" / controller "fail_time" should not reload
+    match (not.channel(), not.payload()) {
+        ("comm_config", _) => true,
+        ("comm_link", nm) => {
+            nm.is_empty() || cfgs.iter().any(|c| c.name() == nm)
+        }
+        ("controller", nm) => {
+            nm.is_empty() || cfgs.iter().any(|c| c.controller() == nm)
+        }
+        ("detector", nm) => {
+            nm.is_empty() || cfgs.iter().any(|c| c.has_detector(nm))
+        }
+        _ => false,
+    }
+}
+
 /// Main entry point
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -124,9 +161,6 @@ async fn main() -> Result<()> {
     let db = Database::new("tms").await?;
     loop {
         poll_comm_links(db.clone()).await?;
-        let mut ticker = interval(Duration::from_secs(60));
-        // apparently, the first tick completes immediately
-        ticker.tick().await;
-        ticker.tick().await;
+        tokio::time::sleep(Duration::from_secs(60)).await;
     }
 }

pollinator/src/bin/pollinator.rs

@@ -158,6 +158,11 @@ impl CommLinkCfg {
         self.long_per_s
     }
 
+    /// Get controller
+    pub fn controller(&self) -> &str {
+        &self.controller
+    }
+
     /// Set user name
     pub fn with_user(mut self, user: &str) -> Self {
         self.user = Some(user.to_string());
@@ -180,6 +185,11 @@ impl CommLinkCfg {
         self.password.as_deref()
     }
 
+    /// Check if configuration has a detector
+    pub fn has_detector(&self, det_id: &str) -> bool {
+        self.detectors.iter().any(|d| d == det_id)
+    }
+
     /// Make detector hashmap
     pub fn make_detectors(&self) -> HashMap<usize, &str> {
         let mut detectors = HashMap::new();

pollinator/src/comm_link.rs

@@ -8,7 +8,6 @@ edition.workspace = true
 [dependencies]
 bb8 = { workspace = true }
 bb8-postgres = { workspace = true }
-futures = { workspace = true }
 hyper = { version = "1.8", features = ["client", "http1"] }
 jiff = "0.2.16"
 log = { workspace = true }

resin/Cargo.toml

@@ -14,13 +14,13 @@
 //
 use crate::database::Database;
 use crate::error::Result;
-use futures::Stream;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
+use tokio::sync::mpsc::{
+    UnboundedReceiver, UnboundedSender, unbounded_channel,
+};
 use tokio_postgres::tls::NoTlsStream;
 use tokio_postgres::{AsyncMessage, Client, Connection, Notification, Socket};
-use tokio_stream::wrappers::UnboundedReceiverStream;
 
 /// Handler for Postgres NOTIFY
 pub struct Notifier {
@@ -77,14 +77,13 @@ impl Database {
     pub async fn notifier(
         self,
         channels: impl Iterator<Item = &str>,
-    ) -> Result<(Notifier, impl Stream<Item = Notification> + Unpin)> {
+    ) -> Result<(Notifier, UnboundedReceiver<Notification>)> {
         let (client, conn) = self.dedicated_client().await?;
         let (tx, rx) = unbounded_channel();
         for channel in channels {
             client.execute(&format!("LISTEN {channel}"), &[]).await?;
         }
         let not = Notifier { client, conn, tx };
-        let stream = Box::pin(UnboundedReceiverStream::new(rx));
-        Ok((not, stream))
+        Ok((not, rx))
     }
 }

resin/src/notify.rs

@@ -1343,6 +1343,8 @@ CREATE TABLE iris.comm_link (
 CREATE FUNCTION iris.comm_link_notify() RETURNS TRIGGER AS
     $comm_link_notify$
 BEGIN
+    -- FIXME: make connected secondary and
+    --        add another channel for connected changes?
     -- all attributes are primary
     NOTIFY comm_link;
     RETURN NULL; -- AFTER trigger return is ignored
@@ -1436,6 +1438,8 @@ CREATE UNIQUE INDEX ctrl_link_drop_idx ON iris.controller
 CREATE FUNCTION iris.controller_notify() RETURNS TRIGGER AS
     $controller_notify$
 BEGIN
+    -- FIXME: make fail_time secondary and
+    --        add another channel for fail_time changes?
     IF (NEW.drop_id IS DISTINCT FROM OLD.drop_id) OR
        (NEW.comm_link IS DISTINCT FROM OLD.comm_link) OR
        (NEW.cabinet_style IS DISTINCT FROM OLD.cabinet_style) OR