wip: notifications applet subscription for receiving notifications

cosmic-applet-notifications/Cargo.toml

@@ -5,8 +5,16 @@ edition = "2021"
 license = "GPL-3.0-or-later"
 
 [dependencies]
+anyhow = "1"
 icon-loader = { version = "0.3.6", features = ["gtk"] }
 libcosmic.workspace = true
 cosmic-time.workspace = true
 cosmic-applet = { path = "../applet" }
-nix = "0.24.1"
+nix = "0.26"
+tokio = { version = "1.24.1", features = ["sync", "rt", "tracing", "macros", "net", "io-util", "io-std"] }
+cosmic-notifications-util = { git = "https://github.com/pop-os/cosmic-notifications" }
+tracing = "0.1"
+ron = "0.8"
+sendfd = { version = "0.4", features = [ "tokio" ] }
+bytemuck = "1"
+tracing-subscriber = "0.3"

cosmic-applet-notifications/src/main.rs

@@ -1,3 +1,5 @@
+mod subscriptions;
+
 use cosmic::iced::wayland::popup::{destroy_popup, get_popup};
 use cosmic::iced::{
     widget::{button, column, row, text, Row, Space},
@@ -13,10 +15,15 @@ use cosmic::widget::{divider, icon};
 use cosmic::Renderer;
 use cosmic::{Element, Theme};
 use cosmic_time::{anim, chain, id, once_cell::sync::Lazy, Instant, Timeline};
-
+use cosmic_notifications_util::AppletEvent;
+use tracing::info;
 use std::process;
 
 pub fn main() -> cosmic::iced::Result {
+    tracing_subscriber::fmt::init();
+
+    info!("Notifications applet");
+
     let helper = CosmicAppletHelper::default();
     Notifications::run(helper.window_settings())
 }
@@ -43,6 +50,7 @@ enum Message {
     Ignore,
     Frame(Instant),
     Theme(Theme),
+    NotificationEvent(AppletEvent)
 }
 
 impl Application for Notifications {
@@ -90,6 +98,7 @@ impl Application for Notifications {
             self.timeline
                 .as_subscription()
                 .map(|(_, now)| Message::Frame(now)),
+            subscriptions::notifications::notifications().map(Message::NotificationEvent)
         ])
     }
 
@@ -130,6 +139,10 @@ impl Application for Notifications {
                 let _ = process::Command::new("cosmic-settings notifications").spawn();
                 Command::none()
             }
+            Message::NotificationEvent(e) => {
+                dbg!(e);
+                Command::none()
+            }
             Message::Ignore => Command::none(),
         }
     }

cosmic-applet-notifications/src/subscriptions/dbus.rs

@@ -0,0 +1 @@
+// TODO connect as a client and send actions / dismissals

cosmic-applet-notifications/src/subscriptions/mod.rs

@@ -0,0 +1 @@
+pub mod notifications;

cosmic-applet-notifications/src/subscriptions/notifications.rs

@@ -0,0 +1,129 @@
+use tokio::{
+    io::{AsyncBufReadExt, BufReader},
+    net::UnixStream,
+    sync::oneshot,
+};
+use tracing::{error, info};
+use cosmic::{
+    iced::{
+        futures::{self, SinkExt},
+        subscription,
+    },
+    iced_futures::Subscription,
+};
+use cosmic_notifications_util::AppletEvent;
+use std::os::unix::io::{FromRawFd, RawFd};
+use sendfd::RecvWithFd;
+
+#[derive(Debug)]
+pub enum State {
+    WaitingForPanel,
+    WaitingForDaemon(UnixStream),
+    WaitingForNotificationEvent(UnixStream),
+    Finished,
+}
+
+pub fn notifications() -> Subscription<AppletEvent> {
+    struct SomeWorker;
+
+    subscription::channel(
+        std::any::TypeId::of::<SomeWorker>(),
+        50,
+        |mut output| async move {
+            let mut state = State::WaitingForPanel;
+
+            loop {
+                match &mut state {
+                    State::WaitingForPanel => {
+                        info!("Waiting for panel to send us a stream");
+
+                        let (tx, rx) = oneshot::channel();
+
+                        std::thread::spawn(move || -> anyhow::Result<()> {
+                            let mut msg = String::new();
+                            std::io::stdin().read_line(&mut msg)?;
+                            let raw_fd =  msg.trim().parse::<RawFd>()?;
+                            if raw_fd == 0 {
+                                anyhow::bail!("Invalid fd received from panel");
+                            }
+                            _ = tx.send(raw_fd);
+                            Ok(())
+                        });
+                        
+                        let Ok(raw_fd) = rx.await else {
+                            error!("Failed to receive raw fd from panel");
+                            state = State::Finished;
+                            continue;
+                        };
+
+                        let stream = unsafe { std::os::unix::net::UnixStream::from_raw_fd(raw_fd) };
+                        let Ok(stream) = UnixStream::from_std(stream) else {
+                            error!("Failed to convert std stream to unix stream");
+                            state = State::Finished;
+                            continue;
+                        };
+                        state = State::WaitingForDaemon(stream);
+
+                    }
+                    State::WaitingForDaemon(stream) => {
+                        info!("Waiting for daemon to send us a stream");
+                        if let Err(err) = stream.readable().await {
+                            error!("Failed to wait for stream to be readable {}", err);
+                            state = State::Finished;
+                            continue;
+                        };
+
+                        // we are expecting a single RawFd from the panel on this stream and the applet id
+                        let mut buf = [0u8; 4];
+                        let mut fd_buf = [0i32; 1];
+
+                        match stream.recv_with_fd(&mut buf, &mut fd_buf) {
+                            Ok((data_cnt, fd_cnt)) => {
+                                if data_cnt != 4 || fd_cnt != 1 {
+                                    error!("Invalid data received from panel");
+                                    state = State::Finished;
+
+                                    continue;
+                                }
+                                let notif_stream = unsafe { std::os::unix::net::UnixStream::from_raw_fd(fd_buf[0]) };
+                                let Ok(notif_stream) = UnixStream::from_std(notif_stream) else {
+                                    error!("Failed to convert raw fd to unix stream");
+                                    state = State::Finished;
+                                    continue;
+                                };
+                                
+                                state = State::WaitingForNotificationEvent(notif_stream);
+                            }
+                            Err(err) => {
+                                error!("Failed to receive fd from panel: {}", err);
+                                state = State::Finished;
+
+                                continue;
+                            }
+                        }
+                    }
+                    State::WaitingForNotificationEvent(stream) => {
+                        let mut reader = BufReader::new(stream);
+                        // todo read messages
+                        let mut request_buf = String::with_capacity(1024);
+                        if let Err(err) = reader.read_line(&mut request_buf).await {
+                            error!("Failed to read line from stream {}", err);
+                            continue;
+                        }
+                        let Ok(event) = ron::de::from_str::<AppletEvent>(request_buf.as_str()) else {
+                            error!("Failed to deserialize event from notification stream");
+                            continue;
+                        };
+
+                        if let Err(err) = output.send(event).await {
+                            error!("Error sending event: {}", err);
+                        }
+                    }
+                    State::Finished => {
+                        let () = futures::future::pending().await;
+                    }
+                }
+            }
+        }
+    )
+}