improv(greeter): use subscription for handling greetd IPC socket

daemon/src/main.rs

@@ -3,7 +3,7 @@ use cosmic_comp_config::CosmicCompConfig;
 use cosmic_config::CosmicConfigEntry;
 use cosmic_greeter_daemon::{UserData, WallpaperData};
 use std::{env, error::Error, fs, future::pending, io, path::Path};
-use zbus::{dbus_interface, ConnectionBuilder, DBusError};
+use zbus::{ConnectionBuilder, DBusError};
 
 //IMPORTANT: this function is critical to the security of this proxy. It must ensure that the
 // callback is executed with the permissions of the specified user id. A good test is to see if
@@ -47,7 +47,7 @@ enum GreeterError {
 
 struct GreeterProxy;
 
-#[dbus_interface(name = "com.system76.CosmicGreeter")]
+#[zbus::interface(name = "com.system76.CosmicGreeter")]
 impl GreeterProxy {
     fn get_user_data(&mut self) -> Result<String, GreeterError> {
         // The pwd::Passwd method is unsafe (but not labelled as such) due to using global state (libc pwent functions).

src/greeter.rs

@@ -1,6 +1,8 @@
 // Copyright 2023 System76 <info@system76.com>
 // SPDX-License-Identifier: GPL-3.0-only
 
+mod ipc;
+
 use cosmic::app::{message, Command, Core, Settings};
 use cosmic::{
     cosmic_config::{self, ConfigSet, CosmicConfigEntry},
@@ -26,10 +28,9 @@ use cosmic::{
 };
 use cosmic_comp_config::CosmicCompConfig;
 use cosmic_greeter_daemon::{UserData, WallpaperData};
-use greetd_ipc::{codec::TokioCodec, AuthMessageType, Request, Response};
+use greetd_ipc::Request;
 use std::{
     collections::HashMap,
-    env,
     error::Error,
     fs, io,
     path::{Path, PathBuf},
@@ -37,8 +38,7 @@ use std::{
     sync::Arc,
     time::{Duration, Instant},
 };
-use tokio::sync::Mutex;
+use tokio::time;
-use tokio::{net::UnixStream, time};
 use wayland_client::{protocol::wl_output::WlOutput, Proxy};
 use zbus::{proxy, Connection};
 
@@ -304,84 +304,6 @@ pub fn main() -> Result<(), Box<dyn Error>> {
     Ok(())
 }
 
-async fn request_message(socket: Arc<Mutex<UnixStream>>, request: Request) -> Message {
-    //TODO: handle errors
-    let response = {
-        let mut socket = socket.lock().await;
-        request.write_to(&mut *socket).await.unwrap();
-        Response::read_from(&mut *socket).await
-    };
-
-    //TODO: handle responses at any time?
-    match response {
-        Ok(response) => {
-            log::info!("{:?}", response);
-            match response {
-                Response::AuthMessage {
-                    auth_message_type,
-                    auth_message,
-                } => match auth_message_type {
-                    AuthMessageType::Secret => {
-                        return Message::Prompt(auth_message, true, Some(String::new()));
-                    }
-                    AuthMessageType::Visible => {
-                        return Message::Prompt(auth_message, false, Some(String::new()));
-                    }
-                    //TODO: treat error type differently?
-                    AuthMessageType::Info | AuthMessageType::Error => {
-                        return Message::Prompt(auth_message, false, None);
-                    }
-                },
-                Response::Error {
-                    error_type: _,
-                    description,
-                } => {
-                    //TODO: use error_type?
-                    match request {
-                        Request::CancelSession => {
-                            // Do not send errors for cancel session to gui
-                            log::warn!("error while cancelling session: {}", description);
-                            // Reconnect to socket
-                            return Message::Reconnect;
-                        }
-                        _ => {
-                    return Message::Error(socket, description);
-                        }
-                    }
-                }
-                Response::Success => match request {
-                    Request::CreateSession { .. } => {
-                        // User has no auth required, proceed to login
-                        return Message::Login(socket);
-                    }
-                    Request::PostAuthMessageResponse { .. } => {
-                        // All auth is completed, proceed to login
-                        return Message::Login(socket);
-                    }
-                    Request::StartSession { .. } => {
-                        // Session has been started, exit greeter
-                        return Message::Exit;
-                    }
-                    Request::CancelSession => {
-                        // Reconnect to socket
-                        return Message::Reconnect;
-                    }
-                },
-            }
-        }
-        Err(err) => log::error!("failed to read socket: {:?}", err),
-    }
-
-    Message::None
-}
-
-fn request_command(socket: Arc<Mutex<UnixStream>>, request: Request) -> Command<Message> {
-    Command::perform(
-        async move { message::app(request_message(socket, request).await) },
-        |x| x,
-    )
-}
-
 #[derive(Clone)]
 pub struct Flags {
     user_datas: Vec<UserData>,
@@ -396,7 +318,7 @@ pub enum SocketState {
     /// Opening GREETD_SOCK
     Pending,
     /// GREETD_SOCK is open
-    Open(Arc<Mutex<UnixStream>>),
+    Open,
     /// No GREETD_SOCK variable set
     NotSet,
     /// Failed to open GREETD_SOCK
@@ -439,34 +361,37 @@ pub enum Dropdown {
 /// Messages that are used specifically by our [`App`].
 #[derive(Clone, Debug)]
 pub enum Message {
-    None,
+    Auth(Option<String>),
-    OutputEvent(OutputEvent, WlOutput),
-    LayerEvent(LayerEvent, SurfaceId),
-    Socket(SocketState),
-    NetworkIcon(Option<&'static str>),
-    PowerInfo(Option<(String, f64)>),
-    Prompt(String, bool, Option<String>),
-    Session(String),
-    Username(String),
-    Auth(Arc<Mutex<UnixStream>>, Option<String>),
-    Login(Arc<Mutex<UnixStream>>),
-    Error(Arc<Mutex<UnixStream>>, String),
     DialogCancel,
     DialogConfirm,
     DropdownToggle(Dropdown),
-    KeyboardLayout(usize),
+    Error(String),
-    Reconnect,
-    Suspend,
-    Restart,
-    Shutdown,
-    Heartbeat,
     Exit,
+    // Sets channel used to communicate with the greetd IPC subscription.
+    GreetdChannel(tokio::sync::mpsc::Sender<Request>),
+    Heartbeat,
+    KeyboardLayout(usize),
+    LayerEvent(LayerEvent, SurfaceId),
+    Login,
+    NetworkIcon(Option<&'static str>),
+    None,
+    OutputEvent(OutputEvent, WlOutput),
+    PowerInfo(Option<(String, f64)>),
+    Prompt(String, bool, Option<String>),
+    Reconnect,
+    Restart,
+    Session(String),
+    Shutdown,
+    Socket(SocketState),
+    Suspend,
+    Username(String),
 }
 
 /// The [`App`] stores application-specific state.
 pub struct App {
     core: Core,
     flags: Flags,
+    greetd_sender: Option<tokio::sync::mpsc::Sender<greetd_ipc::Request>>,
     surface_ids: HashMap<WlOutput, SurfaceId>,
     active_surface_id_opt: Option<SurfaceId>,
     surface_images: HashMap<SurfaceId, widget::image::Handle>,
@@ -487,6 +412,19 @@ pub struct App {
 }
 
 impl App {
+    /// Send a [`Request`] to the greetd IPC subscription.
+    fn send_request(&self, request: Request) -> Command<Message> {
+        if let Some(ref sender) = self.greetd_sender {
+            let sender = sender.clone();
+            return cosmic::command::future(async move {
+                _ = sender.send(request).await;
+                message::none()
+            });
+        }
+
+        Command::none()
+    }
+
     fn set_xkb_config(&self) {
         let user_data = match self
             .flags
@@ -677,9 +615,10 @@ impl cosmic::Application for App {
             .map(|x| x.name.clone())
             .unwrap_or(String::new());
 
-        let mut app = App {
+        let app = App {
             core,
             flags,
+            greetd_sender: None,
             surface_ids: HashMap::new(),
             active_surface_id_opt: None,
             surface_images: HashMap::new(),
@@ -698,8 +637,7 @@ impl cosmic::Application for App {
             dialog_page_opt: None,
             dropdown_opt: None,
         };
-        let command = app.update(Message::Reconnect);
+        (app, Command::none())
-        (app, command)
     }
 
     /// Handle application events here.
@@ -798,14 +736,11 @@ impl cosmic::Application for App {
             Message::Socket(socket_state) => {
                 self.socket_state = socket_state;
                 match &self.socket_state {
-                    SocketState::Open(socket) => {
+                    SocketState::Open => {
                         // When socket is opened, send create session
-                        return Command::batch([request_command(
+                        return self.send_request(Request::CreateSession {
-                            socket.clone(),
+                            username: self.selected_username.clone(),
-                            Request::CreateSession {
+                        });
-                                username: self.selected_username.clone(),
-                            },
-                        )]);
                     }
                     _ => {}
                 }
@@ -845,52 +780,35 @@ impl cosmic::Application for App {
                     self.selected_username = username.clone();
                     self.surface_images.clear();
                     match &self.socket_state {
-                        SocketState::Open(socket) => {
+                        SocketState::Open => {
                             self.prompt_opt = None;
-                            return request_command(socket.clone(), Request::CancelSession);
+                            return self.send_request(Request::CancelSession);
                         }
                         _ => {}
                     }
                 }
             }
-            Message::Auth(socket, response) => {
+            Message::Auth(response) => {
                 self.prompt_opt = None;
                 self.error_opt = None;
-                return request_command(socket, Request::PostAuthMessageResponse { response });
+                return self.send_request(Request::PostAuthMessageResponse { response });
             }
-            Message::Login(socket) => {
+            Message::Login => {
                 self.prompt_opt = None;
                 self.error_opt = None;
                 match self.flags.sessions.get(&self.selected_session).cloned() {
                     Some((cmd, env)) => {
-                        return request_command(socket, Request::StartSession { cmd, env });
+                        return self.send_request(Request::StartSession { cmd, env });
                     }
                     None => todo!("session {:?} not found", self.selected_session),
                 }
             }
-            Message::Error(socket, error) => {
+            Message::Error(error) => {
                 self.error_opt = Some(error);
-                return request_command(socket, Request::CancelSession);
+                return self.send_request(Request::CancelSession);
             }
             Message::Reconnect => {
-                return Command::batch([
+                return self.update_user_config();
-                    self.update_user_config(),
-                    Command::perform(
-                        async {
-                            message::app(Message::Socket(match env::var_os("GREETD_SOCK") {
-                                Some(socket_path) => {
-                                    log::info!("opening {:?}", socket_path);
-                                    match UnixStream::connect(&socket_path).await {
-                                        Ok(socket) => SocketState::Open(Arc::new(socket.into())),
-                                        Err(err) => SocketState::Error(Arc::new(err)),
-                                    }
-                                }
-                                None => SocketState::NotSet,
-                            }))
-                        },
-                        |x| x,
-                    ),
-                ]);
             }
             Message::DialogCancel => {
                 self.dialog_page_opt = None;
@@ -898,33 +816,27 @@ impl cosmic::Application for App {
             Message::DialogConfirm => match self.dialog_page_opt.take() {
                 Some(DialogPage::Restart(_)) => {
                     #[cfg(feature = "logind")]
-                    return Command::perform(
+                    return cosmic::command::future(async move {
-                        async move {
+                        match crate::logind::reboot().await {
-                            match crate::logind::reboot().await {
+                            Ok(()) => (),
-                                Ok(()) => (),
+                            Err(err) => {
-                                Err(err) => {
+                                log::error!("failed to reboot: {:?}", err);
-                                    log::error!("failed to reboot: {:?}", err);
-                                }
                             }
-                            message::none()
+                        }
-                        },
+                        message::none()
-                        |x| x,
+                    });
-                    );
                 }
                 Some(DialogPage::Shutdown(_)) => {
                     #[cfg(feature = "logind")]
-                    return Command::perform(
+                    return cosmic::command::future(async move {
-                        async move {
+                        match crate::logind::power_off().await {
-                            match crate::logind::power_off().await {
+                            Ok(()) => (),
-                                Ok(()) => (),
+                            Err(err) => {
-                                Err(err) => {
+                                log::error!("failed to power off: {:?}", err);
-                                    log::error!("failed to power off: {:?}", err);
-                                }
                             }
-                            message::none()
+                        }
-                        },
+                        message::none()
-                        |x| x,
+                    });
-                    );
                 }
                 None => {}
             },
@@ -946,18 +858,15 @@ impl cosmic::Application for App {
             }
             Message::Suspend => {
                 #[cfg(feature = "logind")]
-                return Command::perform(
+                return cosmic::command::future(async move {
-                    async move {
+                    match crate::logind::suspend().await {
-                        match crate::logind::suspend().await {
+                        Ok(()) => (),
-                            Ok(()) => (),
+                        Err(err) => {
-                            Err(err) => {
+                            log::error!("failed to suspend: {:?}", err);
-                                log::error!("failed to suspend: {:?}", err);
-                            }
                         }
-                        message::none()
+                    }
-                    },
+                    message::none()
-                    |x| x,
+                });
-                );
             }
             Message::Restart => {
                 self.dialog_page_opt = Some(DialogPage::Restart(Instant::now()));
@@ -984,6 +893,9 @@ impl cosmic::Application for App {
                 commands.push(Command::perform(async { process::exit(0) }, |x| x));
                 return Command::batch(commands);
             }
+            Message::GreetdChannel(sender) => {
+                self.greetd_sender = Some(sender);
+            }
         }
         Command::none()
     }
@@ -1191,7 +1103,7 @@ impl cosmic::Application for App {
                 SocketState::Pending => {
                     column = column.push(widget::text("Opening GREETD_SOCK"));
                 }
-                SocketState::Open(socket) => {
+                SocketState::Open => {
                     for user_data in &self.flags.user_datas {
                         if &user_data.name == &self.selected_username {
                             match &user_data.icon_opt {
@@ -1239,10 +1151,7 @@ impl cosmic::Application for App {
                                         .on_input(|value| {
                                             Message::Prompt(prompt.clone(), *secret, Some(value))
                                         })
-                                        .on_submit(Message::Auth(
+                                        .on_submit(Message::Auth(Some(value.clone())));
-                                            socket.clone(),
-                                            Some(value.clone()),
-                                        ));
 
                                 if let Some(text_input_id) = self.text_input_ids.get(&surface_id) {
                                     text_input = text_input.id(text_input_id.clone());
@@ -1255,10 +1164,8 @@ impl cosmic::Application for App {
                                 column = column.push(text_input);
                             }
                             None => {
-                                column = column.push(
+                                column = column
-                                    widget::button("Confirm")
+                                    .push(widget::button("Confirm").on_press(Message::Auth(None)));
-                                        .on_press(Message::Auth(socket.clone(), None)),
-                                );
                             }
                         },
                         None => {}
@@ -1407,6 +1314,7 @@ impl cosmic::Application for App {
                     }
                 },
             ),
+            ipc::subscription(),
             Subscription::batch(extra_suscriptions),
         ])
     }

src/greeter/ipc.rs

@@ -0,0 +1,128 @@
+// Copyright 2024 System76 <info@system76.com>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use super::{Message, SocketState};
+use cosmic::iced::Subscription;
+use futures_util::SinkExt;
+use greetd_ipc::codec::TokioCodec;
+use std::sync::Arc;
+use tokio::net::UnixStream;
+use tokio::sync::mpsc;
+
+pub fn subscription() -> Subscription<Message> {
+    struct GreetdSubscription;
+    cosmic::iced::subscription::channel(
+        std::any::TypeId::of::<GreetdSubscription>(),
+        1,
+        |mut sender| async move {
+            let (tx, mut rx) = mpsc::channel::<greetd_ipc::Request>(1);
+            _ = sender.send(Message::GreetdChannel(tx)).await;
+
+            let socket_path =
+                std::env::var_os("GREETD_SOCK").expect("GREETD_SOCK environment not set");
+
+            loop {
+                _ = sender.send(Message::Reconnect).await;
+
+                let mut stream = match UnixStream::connect(&socket_path).await {
+                    Ok(stream) => stream,
+                    Err(why) => {
+                        _ = sender.send(Message::Socket(SocketState::Error(Arc::new(why))));
+
+                        break;
+                    }
+                };
+
+                _ = sender.send(Message::Socket(SocketState::Open)).await;
+
+                while let Some(request) = rx.recv().await {
+                    if let Err(why) = request.write_to(&mut stream).await {
+                        log::error!("error writing to GREETD_SOCK stream: {why}");
+                        break;
+                    }
+
+                    match greetd_ipc::Response::read_from(&mut stream).await {
+                        Ok(response) => {
+                            match response {
+                                greetd_ipc::Response::AuthMessage {
+                                    auth_message_type,
+                                    auth_message,
+                                } => match auth_message_type {
+                                    greetd_ipc::AuthMessageType::Secret => {
+                                        _ = sender
+                                            .send(Message::Prompt(
+                                                auth_message,
+                                                true,
+                                                Some(String::new()),
+                                            ))
+                                            .await;
+                                    }
+                                    greetd_ipc::AuthMessageType::Visible => {
+                                        _ = sender
+                                            .send(Message::Prompt(
+                                                auth_message,
+                                                false,
+                                                Some(String::new()),
+                                            ))
+                                            .await;
+                                    }
+                                    //TODO: treat error type differently?
+                                    greetd_ipc::AuthMessageType::Info
+                                    | greetd_ipc::AuthMessageType::Error => {
+                                        _ = sender
+                                            .send(Message::Prompt(auth_message, false, None))
+                                            .await;
+                                    }
+                                },
+                                greetd_ipc::Response::Error {
+                                    error_type: _,
+                                    description,
+                                } => {
+                                    //TODO: use error_type?
+                                    match request {
+                                        greetd_ipc::Request::CancelSession => {
+                                            // Do not send errors for cancel session to gui
+                                            log::warn!(
+                                                "error while cancelling session: {}",
+                                                description
+                                            );
+                                            // Reconnect to socket
+                                            _ = break
+                                        }
+                                        _ => {
+                                            _ = sender.send(Message::Error(description)).await;
+                                        }
+                                    }
+                                }
+                                greetd_ipc::Response::Success => match request {
+                                    greetd_ipc::Request::CreateSession { .. } => {
+                                        // User has no auth required, proceed to login
+                                        _ = sender.send(Message::Login).await;
+                                    }
+                                    greetd_ipc::Request::PostAuthMessageResponse { .. } => {
+                                        // All auth is completed, proceed to login
+                                        _ = sender.send(Message::Login).await;
+                                    }
+                                    greetd_ipc::Request::StartSession { .. } => {
+                                        // Session has been started, exit greeter
+                                        _ = sender.send(Message::Exit).await;
+                                    }
+                                    greetd_ipc::Request::CancelSession => {
+                                        // Reconnect to socket
+                                        break;
+                                    }
+                                },
+                            }
+                        }
+                        Err(err) => {
+                            log::error!("failed to read socket: {:?}", err);
+                            break;
+                        }
+                    }
+                }
+            }
+
+            futures_util::future::pending().await
+        },
+    )
+}

src/locker.rs

@@ -480,13 +480,10 @@ impl cosmic::Application for App {
                         Some(value_tx) => {
                             // Clear errors
                             self.error_opt = None;
-                            return Command::perform(
+                            return cosmic::command::future(async move {
-                                async move {
+                                value_tx.send(value).await.unwrap();
-                                    value_tx.send(value).await.unwrap();
+                                Message::Channel(value_tx)
-                                    message::app(Message::Channel(value_tx))
+                            });
-                                },
-                                |x| x,
-                            );
                         }
                         None => log::warn!("tried to submit when value_tx_opt not set"),
                     },
@@ -496,18 +493,15 @@ impl cosmic::Application for App {
             },
             Message::Suspend => {
                 #[cfg(feature = "logind")]
-                return Command::perform(
+                return cosmic::command::future(async move {
-                    async move {
+                    match crate::logind::suspend().await {
-                        match crate::logind::suspend().await {
+                        Ok(()) => message::none(),
-                            Ok(()) => message::none(),
+                        Err(err) => {
-                            Err(err) => {
+                            log::error!("failed to suspend: {:?}", err);
-                                log::error!("failed to suspend: {:?}", err);
+                            message::app(Message::Error(err.to_string()))
-                                message::app(Message::Error(err.to_string()))
-                            }
                         }
-                    },
+                    }
-                    |x| x,
+                });
-                );
             }
             Message::Error(error) => {
                 self.error_opt = Some(error);