dbus: rauc: forward poller status to broker

Author: hnez

src/dbus/rauc.rs

@@ -21,7 +21,7 @@ use anyhow::Result;
 use async_std::stream::StreamExt;
 use async_std::sync::Arc;
 use futures_util::FutureExt;
-use log::{info, warn};
+use log::{error, info, warn};
 use serde::{Deserialize, Serialize};
 
 use super::systemd::{Service, ServiceAction};
@@ -41,21 +41,37 @@ mod demo_mode;
 #[cfg(not(feature = "demo_mode"))]
 mod installer;
 
+#[cfg(not(feature = "demo_mode"))]
+use installer::InstallerProxy;
+
 #[cfg(not(feature = "demo_mode"))]
 mod poller;
 
 #[cfg(not(feature = "demo_mode"))]
-use installer::InstallerProxy;
+use poller::PollerProxy;
 
 #[cfg(feature = "demo_mode")]
 mod imports {
     pub(super) const CHANNELS_DIR: &str = "demo_files/usr/share/tacd/update_channels";
+
+    pub(super) struct PollerProxy<'a> {
+        _dummy: &'a (),
+    }
+
+    impl PollerProxy<'_> {
+        pub async fn new<C>(_conn: C) -> Option<Self> {
+            Some(Self { _dummy: &() })
+        }
+
+        pub async fn poll(&self) -> zbus::Result<()> {
+            Ok(())
+        }
+    }
 }
 
 #[cfg(not(feature = "demo_mode"))]
 mod imports {
     pub(super) use anyhow::bail;
-    pub(super) use log::error;
 
     pub(super) const CHANNELS_DIR: &str = "/usr/share/tacd/update_channels";
 }
@@ -186,11 +202,14 @@ fn would_reboot_into_other_slot(slot_status: &SlotStatus, primary: Option<String
 }
 
 async fn channel_list_update_task(
+    conn: Arc<Connection>,
     reload: Arc<Topic<bool>>,
     enable_polling: Arc<Topic<bool>>,
     channels: Arc<Topic<Channels>>,
     rauc_service: Service,
 ) -> Result<()> {
+    let poller = PollerProxy::new(&conn).await.unwrap();
+
     let (reload_stream, _) = reload.subscribe_unbounded();
     let (mut enable_polling_stream, _) = enable_polling.subscribe_unbounded();
 
@@ -252,6 +271,14 @@ async fn channel_list_update_task(
         }
 
         status_subscription.unsubscribe();
+
+        if enable_polling {
+            info!("Trigger a poll");
+
+            if let Err(err) = poller.poll().await {
+                error!("Failed to poll for updates: {err}");
+            }
+        }
     }
 }
 
@@ -295,6 +322,7 @@ impl Rauc {
         wtb.spawn_task(
             "rauc-channel-list-update",
             channel_list_update_task(
+                Arc::new(Connection),
                 inst.reload.clone(),
                 inst.enable_polling.clone(),
                 inst.channels.clone(),
@@ -448,6 +476,56 @@ impl Rauc {
             Ok(())
         })?;
 
+        let conn_task = conn.clone();
+        let channels = inst.channels.clone();
+
+        // Forward the "Poller::status" property to the broker framework
+        wtb.spawn_task("rauc-forward-poller-status", async move {
+            let proxy = PollerProxy::new(&conn_task).await.unwrap();
+
+            let mut stream = proxy.receive_status_changed().await;
+
+            if let Ok(status) = proxy.status().await {
+                channels.modify(|chs| {
+                    let mut chs = chs?;
+
+                    match chs.update_from_poll_status(status.into()) {
+                        Ok(true) => Some(chs),
+                        Ok(false) => None,
+                        Err(e) => {
+                            warn!("Could not update channel list from poll status: {e}");
+                            None
+                        }
+                    }
+                });
+            }
+
+            while let Some(status) = stream.next().await {
+                let status = match status.get().await {
+                    Ok(status) => status,
+                    Err(e) => {
+                        warn!("Could not get poll status: {e}");
+                        continue;
+                    }
+                };
+
+                channels.modify(|chs| {
+                    let mut chs = chs?;
+
+                    match chs.update_from_poll_status(status.into()) {
+                        Ok(true) => Some(chs),
+                        Ok(false) => None,
+                        Err(e) => {
+                            warn!("Could not update channel list from poll status: {e}");
+                            None
+                        }
+                    }
+                });
+            }
+
+            Ok(())
+        })?;
+
         let conn_task = conn.clone();
         let channels = inst.channels.clone();
         let (mut install_stream, _) = inst.install.clone().subscribe_unbounded();
@@ -503,6 +581,7 @@ impl Rauc {
         wtb.spawn_task(
             "rauc-channel-list-update",
             channel_list_update_task(
+                conn.clone(),
                 inst.reload.clone(),
                 inst.enable_polling.clone(),
                 inst.channels.clone(),

src/dbus/rauc/update_channels.rs

@@ -15,6 +15,8 @@
 // with this program; if not, write to the Free Software Foundation, Inc.,
 // 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
+#[cfg(not(feature = "demo_mode"))]
+use std::convert::TryFrom;
 use std::fs::{read_dir, read_to_string, DirEntry};
 use std::os::unix::ffi::OsStrExt;
 use std::path::Path;
@@ -64,6 +66,34 @@ pub struct ChannelFile {
     pub polling_interval: Option<String>,
 }
 
+#[cfg(not(feature = "demo_mode"))]
+fn zvariant_walk_nested_dicts<'a, T>(map: &'a zvariant::Dict, path: &'a [&'a str]) -> Result<&'a T>
+where
+    &'a T: TryFrom<&'a zvariant::Value<'a>>,
+    <&'a T as TryFrom<&'a zvariant::Value<'a>>>::Error: Into<zvariant::Error>,
+{
+    let (key, rem) = path
+        .split_first()
+        .ok_or_else(|| anyhow!("Got an empty path to walk"))?;
+
+    let value: &zvariant::Value = map
+        .get(key)?
+        .ok_or_else(|| anyhow!("Could not find key \"{key}\" in dict"))?;
+
+    if rem.is_empty() {
+        value.downcast_ref().map_err(|e| {
+            let type_name = std::any::type_name::<T>();
+            anyhow!("Failed to convert value in dictionary for key \"{key}\" to {type_name}: {e}")
+        })
+    } else {
+        let value = value.downcast_ref().map_err(|e| {
+            anyhow!("Failed to convert value in dictionary for key \"{key}\" to a dict: {e}")
+        })?;
+
+        zvariant_walk_nested_dicts(value, rem)
+    }
+}
+
 impl Channel {
     fn from_file(path: &Path) -> Result<Self> {
         let file_name = || {
@@ -177,4 +207,40 @@ impl Channels {
     pub(super) fn primary(&self) -> Option<&Channel> {
         self.0.iter().find(|ch| ch.primary)
     }
+
+    #[cfg(not(feature = "demo_mode"))]
+    fn primary_mut(&mut self) -> Option<&mut Channel> {
+        self.0.iter_mut().find(|ch| ch.primary)
+    }
+
+    #[cfg(not(feature = "demo_mode"))]
+    pub(super) fn update_from_poll_status(&mut self, poll_status: zvariant::Dict) -> Result<bool> {
+        let compatible: &zvariant::Str =
+            zvariant_walk_nested_dicts(&poll_status, &["manifest", "update", "compatible"])?;
+        let version: &zvariant::Str =
+            zvariant_walk_nested_dicts(&poll_status, &["manifest", "update", "version"])?;
+        let newer_than_installed: &bool =
+            zvariant_walk_nested_dicts(&poll_status, &["update-available"])?;
+
+        if let Some(pb) = self.0.iter().find_map(|ch| ch.bundle.as_ref()) {
+            if compatible == pb.compatible.as_str()
+                && version == pb.version.as_str()
+                && *newer_than_installed == pb.newer_than_installed
+            {
+                return Ok(false);
+            }
+        }
+
+        self.0.iter_mut().for_each(|ch| ch.bundle = None);
+
+        if let Some(primary) = self.primary_mut() {
+            primary.bundle = Some(UpstreamBundle {
+                compatible: compatible.as_str().into(),
+                version: version.as_str().into(),
+                newer_than_installed: *newer_than_installed,
+            });
+        }
+
+        Ok(true)
+    }
 }