fix: avoid creating loads of sway/mpd clients

Cargo.lock

@@ -1139,6 +1139,7 @@ dependencies = [
  "gtk",
  "gtk-layer-shell",
  "ksway",
+ "lazy_static",
  "mpd_client",
  "notify",
  "regex",

Cargo.toml

@@ -25,6 +25,7 @@ serde_json = "1.0.82"
 serde_yaml = "0.9.4"
 toml = "0.5.9"
 cornfig = "0.2.0"
+lazy_static = "1.4.0"
 regex = "1.6.0"
 stray = "0.1.1"
 dirs = "4.0.0"

src/main.rs

@@ -11,14 +11,13 @@ mod sway;
 use crate::bar::create_bar;
 use crate::config::{Config, MonitorConfig};
 use crate::style::load_css;
-use crate::sway::{get_client_error, SwayOutput};
+use crate::sway::{get_client, SwayOutput};
 use color_eyre::eyre::Result;
 use color_eyre::Report;
 use dirs::config_dir;
 use gtk::gdk::Display;
 use gtk::prelude::*;
 use gtk::Application;
-use ksway::client::Client;
 use ksway::IpcCommand;
 use std::env;
 use std::process::exit;
@@ -97,14 +96,16 @@ async fn main() -> Result<()> {
 }
 
 fn create_bars(app: &Application, display: &Display, config: &Config) -> Result<()> {
-    let mut sway_client = match Client::connect() {
-        Ok(client) => Ok(client),
-        Err(err) => Err(get_client_error(err)),
-    }?;
+    let outputs = {
+        let sway = get_client();
+        let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
 
-    let outputs = match sway_client.ipc(IpcCommand::GetOutputs) {
+        let outputs = sway.ipc(IpcCommand::GetOutputs);
+
+        match outputs {
             Ok(outputs) => Ok(outputs),
-        Err(err) => Err(get_client_error(err)),
+            Err(err) => Err(err),
+        }
     }?;
 
     let outputs = serde_json::from_slice::<Vec<SwayOutput>>(&outputs)?;

src/modules/focused.rs

@@ -1,14 +1,12 @@
 use crate::icon;
 use crate::modules::{Module, ModuleInfo};
-use crate::sway::{SwayClient, WindowEvent};
+use crate::sway::get_client;
 use color_eyre::Result;
 use glib::Continue;
 use gtk::prelude::*;
 use gtk::{IconTheme, Image, Label, Orientation};
-use ksway::IpcEvent;
 use serde::Deserialize;
 use tokio::task::spawn_blocking;
-use tracing::error;
 
 #[derive(Debug, Deserialize, Clone)]
 pub struct FocusedModule {
@@ -42,24 +40,28 @@ impl Module<gtk::Box> for FocusedModule {
         container.add(&icon);
         container.add(&label);
 
-        let mut sway = SwayClient::connect()?;
-
-        let srx = sway.subscribe(vec![IpcEvent::Window])?;
         let (tx, rx) = glib::MainContext::channel(glib::PRIORITY_DEFAULT);
 
-        let focused = sway
-            .get_open_windows()?
+        let focused = {
+            let sway = get_client();
+            let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
+            sway.get_open_windows()?
                 .into_iter()
-            .find(|node| node.focused);
+                .find(|node| node.focused)
+        };
 
         if let Some(focused) = focused {
             tx.send(focused)?;
         }
 
-        spawn_blocking(move || loop {
-            while let Ok((_, payload)) = srx.try_recv() {
-                match serde_json::from_slice::<WindowEvent>(&payload) {
-                    Ok(payload) => {
+        spawn_blocking(move || {
+            let srx = {
+                let sway = get_client();
+                let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
+                sway.subscribe_window()
+            };
+
+            while let Ok(payload) = srx.recv() {
                 let update = match payload.change.as_str() {
                     "focus" => true,
                     "title" => payload.container.focused,
@@ -71,13 +73,6 @@ impl Module<gtk::Box> for FocusedModule {
                         .expect("Failed to sendf focus update");
                 }
             }
-                    Err(err) => error!("{:?}", err),
-                }
-            }
-
-            if let Err(err) = sway.poll() {
-                error!("{:?}", err);
-            }
         });
 
         {

src/modules/launcher/mod.rs

@@ -5,17 +5,15 @@ use crate::collection::Collection;
 use crate::modules::launcher::item::{ButtonConfig, LauncherItem, LauncherWindow, OpenState};
 use crate::modules::launcher::popup::Popup;
 use crate::modules::{Module, ModuleInfo};
-use crate::sway::{SwayClient, SwayNode, WindowEvent};
+use crate::sway::{get_client, SwayNode};
 use color_eyre::{Report, Result};
 use gtk::prelude::*;
 use gtk::{IconTheme, Orientation};
-use ksway::IpcEvent;
 use serde::Deserialize;
 use std::rc::Rc;
 use tokio::spawn;
 use tokio::sync::mpsc;
 use tokio::task::spawn_blocking;
-use tracing::error;
 
 #[derive(Debug, Deserialize, Clone)]
 pub struct LauncherModule {
@@ -210,8 +208,6 @@ impl Module<gtk::Box> for LauncherModule {
             icon_theme.set_custom_theme(Some(&theme));
         }
 
-        let mut sway = SwayClient::connect()?;
-
         let popup = Popup::new(
             "popup-launcher",
             info.app,
@@ -237,29 +233,29 @@ impl Module<gtk::Box> for LauncherModule {
             button_config,
         );
 
-        let open_windows = sway.get_open_windows()?;
+        let open_windows = {
+            let sway = get_client();
+            let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
+            sway.get_open_windows()
+        }?;
 
         for window in open_windows {
             launcher.add_window(window);
         }
 
-        let srx = sway.subscribe(vec![IpcEvent::Window])?;
         let (tx, rx) = glib::MainContext::channel(glib::PRIORITY_DEFAULT);
 
-        spawn_blocking(move || loop {
-            while let Ok((_, payload)) = srx.try_recv() {
-                match serde_json::from_slice::<WindowEvent>(&payload) {
-                    Ok(payload) => {
+        spawn_blocking(move || {
+            let srx = {
+                let sway = get_client();
+                let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
+                sway.subscribe_window()
+            };
+
+            while let Ok(payload) = srx.recv() {
                 tx.send(payload)
                     .expect("Failed to send window event payload");
             }
-                    Err(err) => error!("{:?}", err),
-                }
-            }
-
-            if let Err(err) = sway.poll() {
-                error!("{:?}", err);
-            }
         });
 
         {
@@ -278,14 +274,15 @@ impl Module<gtk::Box> for LauncherModule {
         }
 
         spawn(async move {
-            let mut sway = SwayClient::connect()?;
+            let sway = get_client();
+
             while let Some(event) = ui_rx.recv().await {
                 let selector = match event {
                     FocusEvent::AppId(app_id) => format!("[app_id={}]", app_id),
                     FocusEvent::Class(class) => format!("[class={}]", class),
                     FocusEvent::ConId(id) => format!("[con_id={}]", id),
                 };
-
+                let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
                 sway.run(format!("{} focus", selector))?;
             }
 

src/modules/mpd/client.rs

@@ -1,55 +1,67 @@
+use lazy_static::lazy_static;
 use mpd_client::commands::responses::Status;
 use mpd_client::raw::MpdProtocolError;
 use mpd_client::{Client, Connection};
+use std::collections::HashMap;
 use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::net::{TcpStream, UnixStream};
-use tokio::spawn;
+use tokio::sync::Mutex;
 use tokio::time::sleep;
 
-pub async fn wait_for_connection(
-    hosts: Vec<String>,
+lazy_static! {
+    static ref CLIENTS: Arc<Mutex<HashMap<String, Arc<Client>>>> =
+        Arc::new(Mutex::new(HashMap::new()));
+}
+
+pub async fn get_connection(host: &str) -> Option<Arc<Client>> {
+    let mut clients = CLIENTS.lock().await;
+
+    match clients.get(host) {
+        Some(client) => Some(Arc::clone(client)),
+        None => {
+            let client = wait_for_connection(host, Duration::from_secs(5), None).await?;
+            let client = Arc::new(client);
+            clients.insert(host.to_string(), Arc::clone(&client));
+            Some(client)
+        }
+    }
+}
+
+async fn wait_for_connection(
+    host: &str,
     interval: Duration,
     max_retries: Option<usize>,
 ) -> Option<Client> {
     let mut retries = 0;
-
-    spawn(async move {
     let max_retries = max_retries.unwrap_or(usize::MAX);
+
     loop {
         if retries == max_retries {
             break None;
         }
 
-            if let Some(conn) = try_get_mpd_conn(&hosts).await {
+        if let Some(conn) = try_get_mpd_conn(host).await {
             break Some(conn.0);
         }
 
         retries += 1;
         sleep(interval).await;
     }
-    })
-    .await
-    .expect("Error occurred while handling tasks")
 }
 
 /// Cycles through each MPD host and
 /// returns the first one which connects,
 /// or none if there are none
-async fn try_get_mpd_conn(hosts: &[String]) -> Option<Connection> {
-    for host in hosts {
+async fn try_get_mpd_conn(host: &str) -> Option<Connection> {
     let connection = if is_unix_socket(host) {
         connect_unix(host).await
     } else {
         connect_tcp(host).await
     };
 
-        if let Ok(connection) = connection {
-            return Some(connection);
-        }
-    }
-
-    None
+    connection.ok()
 }
 
 fn is_unix_socket(host: &str) -> bool {

src/modules/mpd/mod.rs

@@ -2,7 +2,7 @@ mod client;
 mod popup;
 
 use self::popup::Popup;
-use crate::modules::mpd::client::{get_duration, get_elapsed, wait_for_connection};
+use crate::modules::mpd::client::{get_connection, get_duration, get_elapsed};
 use crate::modules::mpd::popup::{MpdPopup, PopupEvent};
 use crate::modules::{Module, ModuleInfo};
 use color_eyre::Result;
@@ -120,7 +120,7 @@ impl Module<Button> for MpdModule {
         let host = self.host.clone();
         let host2 = self.host.clone();
         spawn(async move {
-            let client = wait_for_connection(vec![host], Duration::from_secs(1), None)
+            let client = get_connection(&host)
                 .await
                 .expect("Unexpected error when trying to connect to MPD server");
 
@@ -145,7 +145,7 @@ impl Module<Button> for MpdModule {
         });
 
         spawn(async move {
-            let client = wait_for_connection(vec![host2], Duration::from_secs(1), None)
+            let client = get_connection(&host2)
                 .await
                 .expect("Unexpected error when trying to connect to MPD server");
 

src/modules/workspaces.rs

@@ -1,15 +1,15 @@
 use crate::modules::{Module, ModuleInfo};
-use crate::sway::{SwayClient, Workspace, WorkspaceEvent};
+use crate::sway::{get_client, Workspace};
 use color_eyre::{Report, Result};
 use gtk::prelude::*;
 use gtk::{Button, Orientation};
-use ksway::{IpcCommand, IpcEvent};
+use ksway::IpcCommand;
 use serde::Deserialize;
 use std::collections::HashMap;
 use tokio::spawn;
 use tokio::sync::mpsc;
 use tokio::task::spawn_blocking;
-use tracing::error;
+use tracing::{debug, trace};
 
 #[derive(Debug, Deserialize, Clone)]
 pub struct WorkspacesModule {
@@ -47,11 +47,12 @@ impl Workspace {
 
 impl Module<gtk::Box> for WorkspacesModule {
     fn into_widget(self, info: &ModuleInfo) -> Result<gtk::Box> {
-        let mut sway = SwayClient::connect()?;
-
         let container = gtk::Box::new(Orientation::Horizontal, 0);
 
         let workspaces = {
+            trace!("Getting current workspaces");
+            let sway = get_client();
+            let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
             let raw = sway.ipc(IpcCommand::GetWorkspaces)?;
             let workspaces = serde_json::from_slice::<Vec<Workspace>>(&raw)?;
 
@@ -77,19 +78,19 @@ impl Module<gtk::Box> for WorkspacesModule {
             button_map.insert(workspace.name, item);
         }
 
-        let srx = sway.subscribe(vec![IpcEvent::Workspace])?;
         let (tx, rx) = glib::MainContext::channel(glib::PRIORITY_DEFAULT);
 
-        spawn_blocking(move || loop {
-            while let Ok((_, payload)) = srx.try_recv() {
-                match serde_json::from_slice::<WorkspaceEvent>(&payload) {
-                    Ok(payload) => tx.send(payload).expect("Failed to send workspace event"),
-                    Err(err) => error!("{:?}", err),
-                }
-            }
+        spawn_blocking(move || {
+            trace!("Starting workspace event listener task");
+            let srx = {
+                let sway = get_client();
+                let mut sway = sway.lock().expect("Failed to get lock on Sway IPC client");
 
-            if let Err(err) = sway.poll() {
-                error!("{:?}", err);
+                sway.subscribe_workspace()
+            };
+
+            while let Ok(payload) = srx.recv() {
+                tx.send(payload).expect("Failed to send workspace event");
             }
         });
 
@@ -150,8 +151,12 @@ impl Module<gtk::Box> for WorkspacesModule {
         }
 
         spawn(async move {
-            let mut sway = SwayClient::connect()?;
+            trace!("Setting up UI event handler");
+            let sway = get_client();
             while let Some(name) = ui_rx.recv().await {
+                let mut sway = sway
+                    .lock()
+                    .expect("Failed to get write lock on Sway IPC client");
                 sway.run(format!("workspace {}", name))?;
             }
 

src/sway/mod.rs

@@ -1,17 +1,22 @@
 use color_eyre::{Report, Result};
+use crossbeam_channel::Receiver;
 use ksway::{Error, IpcCommand, IpcEvent};
+use lazy_static::lazy_static;
 use serde::Deserialize;
+use std::sync::{Arc, Mutex};
+use tokio::spawn;
+use tracing::{debug, info, trace};
 
 pub mod node;
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Debug, Clone)]
 pub struct WorkspaceEvent {
     pub change: String,
     pub old: Option<Workspace>,
     pub current: Option<Workspace>,
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Debug, Clone)]
 pub struct Workspace {
     pub name: String,
     pub focused: bool,
@@ -19,13 +24,13 @@ pub struct Workspace {
     pub output: String,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct WindowEvent {
     pub change: String,
     pub container: SwayNode,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct SwayNode {
     #[serde(rename = "type")]
     pub node_type: String,
@@ -40,7 +45,7 @@ pub struct SwayNode {
     pub window_properties: Option<WindowProperties>,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct WindowProperties {
     pub class: Option<String>,
 }
@@ -52,25 +57,79 @@ pub struct SwayOutput {
 
 pub struct SwayClient {
     client: ksway::Client,
+
+    workspace_bc: Arc<Mutex<UnboundedBroadcast<WorkspaceEvent>>>,
+    window_bc: Arc<Mutex<UnboundedBroadcast<WindowEvent>>>,
 }
 
 impl SwayClient {
-    pub(crate) fn run(&mut self, cmd: String) -> Result<Vec<u8>> {
-        match self.client.run(cmd) {
-            Ok(res) => Ok(res),
-            Err(err) => Err(get_client_error(err)),
-        }
-    }
-}
-
-impl SwayClient {
-    pub fn connect() -> Result<Self> {
+    fn connect() -> Result<Self> {
         let client = match ksway::Client::connect() {
             Ok(client) => Ok(client),
             Err(err) => Err(get_client_error(err)),
         }?;
+        info!("Sway IPC client connected");
 
-        Ok(Self { client })
+        let workspace_bc = Arc::new(Mutex::new(UnboundedBroadcast::new()));
+        let window_bc = Arc::new(Mutex::new(UnboundedBroadcast::new()));
+
+        let workspace_bc2 = workspace_bc.clone();
+        let window_bc2 = window_bc.clone();
+        spawn(async move {
+            let mut sub_client = match ksway::Client::connect() {
+                Ok(client) => Ok(client),
+                Err(err) => Err(get_client_error(err)),
+            }
+            .expect("Failed to connect to Sway IPC server");
+            info!("Sway IPC subscription client connected");
+
+            let event_types = vec![IpcEvent::Window, IpcEvent::Workspace];
+            let rx = match sub_client.subscribe(event_types) {
+                Ok(res) => Ok(res),
+                Err(err) => Err(get_client_error(err)),
+            }
+            .expect("Failed to subscribe to Sway IPC server");
+
+            loop {
+                while let Ok((ev_type, payload)) = rx.try_recv() {
+                    debug!("Received sway event {:?}", ev_type);
+                    match ev_type {
+                        IpcEvent::Workspace => {
+                            let json = serde_json::from_slice::<WorkspaceEvent>(&payload).expect(
+                                "Received invalid workspace event payload from Sway IPC server",
+                            );
+                            workspace_bc
+                                .lock()
+                                .expect("Failed to get lock on workspace event bus")
+                                .send(json)
+                                .expect("Failed to broadcast workspace event");
+                        }
+                        IpcEvent::Window => {
+                            let json = serde_json::from_slice::<WindowEvent>(&payload).expect(
+                                "Received invalid window event payload from Sway IPC server",
+                            );
+                            window_bc
+                                .lock()
+                                .expect("Failed to get lock on window event bus")
+                                .send(json)
+                                .expect("Failed to broadcast window event");
+                        }
+                        _ => {}
+                    }
+                }
+                match sub_client.poll() {
+                    Ok(()) => Ok(()),
+                    Err(err) => Err(get_client_error(err)),
+                }
+                .expect("Failed to poll Sway IPC client");
+            }
+        });
+
+        Ok(Self {
+            client,
+            workspace_bc: workspace_bc2,
+            window_bc: window_bc2,
+        })
     }
 
     pub fn ipc(&mut self, command: IpcCommand) -> Result<Vec<u8>> {
@@ -80,21 +139,28 @@ impl SwayClient {
         }
     }
 
-    pub fn subscribe(
-        &mut self,
-        event_types: Vec<IpcEvent>,
-    ) -> Result<crossbeam_channel::Receiver<(IpcEvent, Vec<u8>)>> {
-        match self.client.subscribe(event_types) {
+    pub(crate) fn run(&mut self, cmd: String) -> Result<Vec<u8>> {
+        debug!("Sending command: {}", cmd);
+        match self.client.run(cmd) {
             Ok(res) => Ok(res),
             Err(err) => Err(get_client_error(err)),
         }
     }
 
-    pub fn poll(&mut self) -> Result<()> {
-        match self.client.poll() {
-            Ok(()) => Ok(()),
-            Err(err) => Err(get_client_error(err)),
+    pub fn subscribe_workspace(&mut self) -> Receiver<WorkspaceEvent> {
+        trace!("Adding new workspace subscriber");
+        self.workspace_bc
+            .lock()
+            .expect("Failed to get lock on workspace event bus")
+            .subscribe()
     }
+
+    pub fn subscribe_window(&mut self) -> Receiver<WindowEvent> {
+        trace!("Adding new window subscriber");
+        self.window_bc
+            .lock()
+            .expect("Failed to get lock on window event bus")
+            .subscribe()
     }
 }
 
@@ -107,3 +173,43 @@ pub fn get_client_error(error: Error) -> Report {
         Error::Io(err) => Report::new(err),
     }
 }
+
+lazy_static! {
+    static ref CLIENT: Arc<Mutex<SwayClient>> = {
+        let client = SwayClient::connect();
+        match client {
+            Ok(client) => Arc::new(Mutex::new(client)),
+            Err(err) => panic!("{:?}", err),
+        }
+    };
+}
+
+pub fn get_client() -> Arc<Mutex<SwayClient>> {
+    Arc::clone(&CLIENT)
+}
+
+pub struct UnboundedBroadcast<T> {
+    channels: Vec<crossbeam_channel::Sender<T>>,
+}
+
+impl<T: 'static + Clone + Send + Sync> UnboundedBroadcast<T> {
+    pub const fn new() -> Self {
+        Self { channels: vec![] }
+    }
+
+    pub fn subscribe(&mut self) -> Receiver<T> {
+        let (tx, rx) = crossbeam_channel::unbounded();
+
+        self.channels.push(tx);
+
+        rx
+    }
+
+    pub fn send(&self, message: T) -> Result<(), crossbeam_channel::SendError<T>> {
+        for c in &self.channels {
+            c.send(message.clone())?;
+        }
+
+        Ok(())
+    }
+}