wip output events rework

src/cached_broadcast.rs

@@ -0,0 +1,145 @@
+use crate::{arc_rw, read_lock, send_async, write_lock};
+use std::fmt::Debug;
+use std::ops::Deref;
+use std::sync::{Arc, RwLock};
+// use std::thread::sleep;
+use std::time::Duration;
+use tokio::spawn;
+use tokio::sync::mpsc;
+use tokio::task::spawn_blocking;
+use tokio::time::sleep;
+use tracing::trace;
+
+pub trait Cacheable: Debug + Clone + Send + Sync {
+    type Key: Debug + Clone + Send + Sync + Eq;
+
+    fn get_key(&self) -> Self::Key;
+}
+
+pub type Sender<T> = mpsc::Sender<Event<T>>;
+pub type Receiver<T> = mpsc::Receiver<Event<T>>;
+
+pub struct CachedBroadcastChannel<T>
+where
+    T: Cacheable,
+{
+    capacity: usize,
+    data: Vec<T>,
+    channels: Arc<RwLock<Vec<mpsc::Sender<Event<T>>>>>,
+    base_tx: mpsc::Sender<Event<T>>,
+}
+
+#[derive(Debug, Clone)]
+pub enum Event<T>
+where
+    T: Cacheable,
+{
+    Add(T),
+    Remove(T::Key),
+    Replace(T::Key, T),
+}
+
+impl<T> CachedBroadcastChannel<T>
+where
+    T: Cacheable + 'static,
+{
+    pub fn new(capacity: usize) -> Self {
+        let (tx, rx) = mpsc::channel::<Event<T>>(capacity);
+        let mut rx = DropDetector(rx);
+
+        // spawn_blocking(move || loop {
+        //     let ev = rx.0.try_recv();
+        //     println!("{ev:?}");
+        //     sleep(Duration::from_secs(1))
+        // });
+
+        let channels = arc_rw!(Vec::<Sender<T>>::new());
+
+        let channels = Arc::clone(&channels);
+        spawn(async move {
+            println!("hello");
+
+            while let Some(event) = rx.0.recv().await {
+                println!("ev");
+                // trace!("{event:?}");
+                // let iter = read_lock!(channels).clone().into_iter();
+                // for channel in iter {
+                //     send_async!(channel, event.clone());
+                // }
+            }
+            println!("goodbye");
+        });
+
+        Self {
+            capacity,
+            data: vec![],
+            channels,
+            base_tx: tx,
+        }
+    }
+
+    pub async fn send(&mut self, event: Event<T>) {
+        match event.clone() {
+            Event::Add(data) => {
+                self.data.push(data);
+            }
+            Event::Remove(key) => {
+                let Some(index) = self.data.iter().position(|t| t.get_key() == key) else {
+                    return;
+                };
+                self.data.remove(index);
+            }
+            Event::Replace(key, data) => {
+                let Some(index) = self.data.iter().position(|t| t.get_key() == key) else {
+                    return;
+                };
+                let _ = std::mem::replace(&mut self.data[index], data);
+            }
+        }
+
+        send_async!(self.base_tx, event);
+
+        // let mut closed = vec![];
+        // for (i, channel) in read_lock!(self.channels).iter().enumerate() {
+        //     if channel.is_closed() {
+        //         closed.push(i);
+        //     } else {
+        //         send_async!(channel, event.clone());
+        //     }
+        // }
+        //
+        // for channel in closed.into_iter().rev() {
+        //     write_lock!(self.channels).remove(channel);
+        // }
+    }
+
+    pub fn sender(&self) -> mpsc::Sender<Event<T>> {
+        self.base_tx.clone()
+    }
+
+    pub fn receiver(&mut self) -> mpsc::Receiver<Event<T>> {
+        let (tx, rx) = mpsc::channel(self.capacity);
+        write_lock!(self.channels).push(tx);
+
+        rx
+    }
+
+    pub fn data(&self) -> &Vec<T> {
+        &self.data
+    }
+}
+
+#[derive(Debug)]
+struct DropDetector<T>(T);
+
+impl<T> Drop for DropDetector<T> {
+    fn drop(&mut self) {
+        println!("DROPPED")
+    }
+}
+
+impl<T: Cacheable> Drop for CachedBroadcastChannel<T> {
+    fn drop(&mut self) {
+        println!("Channel DROPPED")
+    }
+}

src/clients/wayland/client.rs

@@ -2,8 +2,9 @@ use super::wlr_foreign_toplevel::handle::ToplevelHandle;
 use super::wlr_foreign_toplevel::manager::ToplevelManagerState;
 use super::wlr_foreign_toplevel::ToplevelEvent;
 use super::Environment;
+use crate::cached_broadcast::CachedBroadcastChannel;
 use crate::error::ERR_CHANNEL_RECV;
-use crate::send;
+use crate::{cached_broadcast, send};
 use cfg_if::cfg_if;
 use color_eyre::Report;
 use smithay_client_toolkit::output::{OutputInfo, OutputState};
@@ -31,11 +32,8 @@ cfg_if! {
 
 #[derive(Debug)]
 pub enum Request {
-    /// Sends a request for all the outputs.
-    /// These are then sent on the `output` channel.
-    Outputs,
     /// Sends a request for all the seats.
-    /// These are then sent ont the `seat` channel.
+    /// These are then sent on the `seat` channel.
     Seats,
     /// Sends a request for all the toplevels.
     /// These are then sent on the `toplevel_init` channel.
@@ -53,8 +51,10 @@ pub enum Request {
 
 pub struct WaylandClient {
     // External channels
+    output_channel: CachedBroadcastChannel<OutputInfo>,
     toplevel_tx: broadcast::Sender<ToplevelEvent>,
     _toplevel_rx: broadcast::Receiver<ToplevelEvent>,
+
     #[cfg(feature = "clipboard")]
     clipboard_tx: broadcast::Sender<Arc<ClipboardItem>>,
     #[cfg(feature = "clipboard")]
@@ -62,7 +62,6 @@ pub struct WaylandClient {
 
     // Internal channels
     toplevel_init_rx: mpsc::Receiver<HashMap<usize, ToplevelHandle>>,
-    output_rx: mpsc::Receiver<Vec<OutputInfo>>,
     seat_rx: mpsc::Receiver<Vec<WlSeat>>,
     #[cfg(feature = "clipboard")]
     clipboard_init_rx: mpsc::Receiver<Option<Arc<ClipboardItem>>>,
@@ -74,10 +73,14 @@ impl WaylandClient {
     pub(super) fn new() -> Self {
         let (toplevel_tx, toplevel_rx) = broadcast::channel(32);
 
+        let mut output_channel = CachedBroadcastChannel::new(8);
+        let output_tx = output_channel.sender();
+
+        let tx2 = output_tx.clone();
+
         let (toplevel_init_tx, toplevel_init_rx) = mpsc::channel();
         #[cfg(feature = "clipboard")]
         let (clipboard_init_tx, clipboard_init_rx) = mpsc::channel();
-        let (output_tx, output_rx) = mpsc::channel();
         let (seat_tx, seat_rx) = mpsc::channel();
 
         let toplevel_tx2 = toplevel_tx.clone();
@@ -99,6 +102,7 @@ impl WaylandClient {
 
             let conn =
                 Connection::connect_to_env().expect("Failed to connect to Wayland compositor");
+
             let (globals, queue) =
                 registry_queue_init(&conn).expect("Failed to retrieve Wayland globals");
 
@@ -139,6 +143,7 @@ impl WaylandClient {
                 handles: HashMap::new(),
                 #[cfg(feature = "clipboard")]
                 clipboard: crate::arc_mut!(None),
+                output_tx,
                 toplevel_tx,
                 #[cfg(feature = "clipboard")]
                 clipboard_tx,
@@ -156,11 +161,6 @@ impl WaylandClient {
                     trace!("{event:?}");
                     match event {
                         Event::Msg(Request::Roundtrip) => debug!("Received refresh event"),
-                        Event::Msg(Request::Outputs) => {
-                            trace!("Received get outputs request");
-
-                            send!(output_tx, env.output_info());
-                        }
                         Event::Msg(Request::Seats) => {
                             trace!("Receive get seats request");
                             send!(seat_tx, env.seats.clone());
@@ -196,12 +196,12 @@ impl WaylandClient {
         });
 
         Self {
+            output_channel,
             toplevel_tx,
             _toplevel_rx: toplevel_rx,
             toplevel_init_rx,
             #[cfg(feature = "clipboard")]
             clipboard_init_rx,
-            output_rx,
             seat_rx,
             #[cfg(feature = "clipboard")]
             clipboard_tx,
@@ -242,6 +242,10 @@ impl WaylandClient {
         (rx, data)
     }
 
+    pub fn subscribe_outputs(&mut self) -> cached_broadcast::Receiver<OutputInfo> {
+        self.output_channel.receiver()
+    }
+
     /// Force a roundtrip on the wayland connection,
     /// flushing any queued events and immediately receiving any new ones.
     pub fn roundtrip(&self) {
@@ -249,11 +253,13 @@ impl WaylandClient {
         send!(self.request_tx, Request::Roundtrip);
     }
 
-    pub fn get_outputs(&self) -> Vec<OutputInfo> {
+    /// Gets a list of all outputs.
-        trace!("Sending get outputs request");
+    ///
-
+    /// This should only be used in a scenario
-        send!(self.request_tx, Request::Outputs);
+    /// where you need a snapshot of outputs at the current time.
-        self.output_rx.recv().expect(ERR_CHANNEL_RECV)
+    /// Prefer to listen to output events with `subscribe_output` where possible.
+    pub fn get_outputs(&self) -> &Vec<OutputInfo> {
+        self.output_channel.data()
     }
 
     pub fn get_seats(&self) -> Vec<WlSeat> {

src/clients/wayland/mod.rs

@@ -6,10 +6,10 @@ mod wl_seat;
 mod wlr_foreign_toplevel;
 
 use self::wlr_foreign_toplevel::manager::ToplevelManagerState;
-use crate::{arc_mut, delegate_foreign_toplevel_handle, delegate_foreign_toplevel_manager};
+use crate::{arc_mut, cached_broadcast, delegate_foreign_toplevel_handle, delegate_foreign_toplevel_manager};
 use cfg_if::cfg_if;
 use lazy_static::lazy_static;
-use smithay_client_toolkit::output::OutputState;
+use smithay_client_toolkit::output::{OutputInfo, OutputState};
 use smithay_client_toolkit::reexports::calloop::LoopHandle;
 use smithay_client_toolkit::registry::{ProvidesRegistryState, RegistryState};
 use smithay_client_toolkit::seat::SeatState;
@@ -65,6 +65,7 @@ pub struct Environment {
     #[cfg(feature = "clipboard")]
     clipboard: Arc<Mutex<Option<Arc<ClipboardItem>>>>,
 
+    output_tx: cached_broadcast::Sender<OutputInfo>,
     toplevel_tx: broadcast::Sender<ToplevelEvent>,
     #[cfg(feature = "clipboard")]
     clipboard_tx: broadcast::Sender<Arc<ClipboardItem>>,

src/clients/wayland/wl_output.rs

@@ -1,4 +1,6 @@
 use super::Environment;
+use crate::cached_broadcast::Cacheable;
+use crate::{cached_broadcast, try_send};
 use smithay_client_toolkit::output::{OutputHandler, OutputInfo, OutputState};
 use tracing::debug;
 use wayland_client::protocol::wl_output;
@@ -31,9 +33,12 @@ impl OutputHandler for Environment {
         &mut self,
         _conn: &Connection,
         _qh: &QueueHandle<Self>,
-        _output: wl_output::WlOutput,
+        output: wl_output::WlOutput,
     ) {
         debug!("Handler received new output");
+        if let Some(info) = self.output_state.info(&output) {
+            try_send!(self.output_tx, cached_broadcast::Event::Add(info));
+        };
     }
 
     fn update_output(
@@ -42,14 +47,26 @@ impl OutputHandler for Environment {
         _qh: &QueueHandle<Self>,
         _output: wl_output::WlOutput,
     ) {
+        debug!("Handle received output update");
     }
 
     fn output_destroyed(
         &mut self,
         _conn: &Connection,
         _qh: &QueueHandle<Self>,
-        _output: wl_output::WlOutput,
+        output: wl_output::WlOutput,
     ) {
         debug!("Handle received output destruction");
+        if let Some(info) = self.output_state.info(&output) {
+            try_send!(self.output_tx, cached_broadcast::Event::Remove(info.id));
+        };
+    }
+}
+
+impl Cacheable for OutputInfo {
+    type Key = u32;
+
+    fn get_key(&self) -> Self::Key {
+        self.id
     }
 }

src/ipc/server.rs

@@ -18,7 +18,7 @@ use crate::ipc::{Command, Response};
 use crate::ironvar::get_variable_manager;
 use crate::modules::PopupButton;
 use crate::style::load_css;
-use crate::{read_lock, send_async, try_send, write_lock, GlobalState};
+use crate::{await_sync, read_lock, send_async, try_send, write_lock, GlobalState};
 
 use super::Ipc;
 
@@ -123,14 +123,7 @@ impl Ipc {
                 Response::Ok
             }
             Command::Reload => {
-                info!("Closing existing bars");
+                await_sync(async move { crate::reload(application, global_state).await }).unwrap();
-                let windows = application.windows();
-                for window in windows {
-                    window.close();
-                }
-
-                crate::load_interface(application, global_state);
-
                 Response::Ok
             }
             Command::Set { key, value } => {

src/main.rs

@@ -14,10 +14,12 @@ use clap::Parser;
 use color_eyre::eyre::Result;
 use color_eyre::Report;
 use dirs::config_dir;
-use gtk::gdk::Display;
+use gtk::gdk::{Display, Monitor};
 use gtk::prelude::*;
 use gtk::Application;
+use smithay_client_toolkit::output::OutputInfo;
 use tokio::runtime::Handle;
+use tokio::spawn;
 use tokio::task::{block_in_place, spawn_blocking};
 use tracing::{debug, error, info, warn};
 use universal_config::ConfigLoader;
@@ -25,6 +27,8 @@ use universal_config::ConfigLoader;
 use clients::wayland;
 
 use crate::bar::create_bar;
+use crate::bridge_channel::BridgeChannel;
+use crate::cached_broadcast::Event;
 use crate::config::{Config, MonitorConfig};
 use crate::error::ExitCode;
 use crate::global_state::GlobalState;
@@ -32,6 +36,7 @@ use crate::style::load_css;
 
 mod bar;
 mod bridge_channel;
+mod cached_broadcast;
 #[cfg(feature = "cli")]
 mod cli;
 mod clients;
@@ -84,19 +89,22 @@ async fn run_with_args(global_state: Rc<RefCell<GlobalState>>) {
                 Err(err) => error!("{err:?}"),
             };
         }
-        None => start_ironbar(global_state),
+        None => start_ironbar(global_state).await,
     }
 }
 
-fn start_ironbar(global_state: Rc<RefCell<GlobalState>>) {
+async fn start_ironbar(global_state: Rc<RefCell<GlobalState>>) {
     info!("Ironbar version {}", VERSION);
     info!("Starting application");
 
     let app = Application::builder().application_id(GTK_APP_ID).build();
-    let _ = wayland::get_client(); // force-init
+
+    let output_bridge = BridgeChannel::<Event<OutputInfo>>::new();
+    let output_tx = output_bridge.create_sender();
 
     let running = Rc::new(Cell::new(false));
 
+    let global_state2 = global_state.clone();
     app.connect_activate(move |app| {
         if running.get() {
             info!("Ironbar already running, returning");
@@ -107,13 +115,11 @@ fn start_ironbar(global_state: Rc<RefCell<GlobalState>>) {
 
         cfg_if! {
             if #[cfg(feature = "ipc")] {
-                let ipc = ipc::Ipc::new(global_state.clone());
+                let ipc = ipc::Ipc::new(global_state2.clone());
                 ipc.start(app);
             }
         }
 
-        load_interface(app, &global_state);
-
         let style_path = env::var("IRONBAR_CSS").ok().map_or_else(
             || {
                 config_dir().map_or_else(
@@ -147,27 +153,131 @@ fn start_ironbar(global_state: Rc<RefCell<GlobalState>>) {
             exit(0);
         });
 
-        ctrlc::set_handler(move || tx.send(()).expect("Could not send signal on channel."))
+        let wc = wayland::get_client();
-            .expect("Error setting Ctrl-C handler");
+
+        let output_tx = output_tx.clone();
+        let mut output_rx = lock!(wc).subscribe_outputs();
+
+        spawn(async move {
+            while let Some(event) = output_rx.recv().await {
+                try_send!(output_tx.clone(), event);
+            }
+        });
+
+        ctrlc::set_handler(move || send!(tx, ())).expect("Error setting Ctrl-C handler");
     });
 
+    let config = load_config();
+
+    {
+        let app = app.clone();
+        let global_state = global_state.clone();
+
+        output_bridge.recv(move |event: cached_broadcast::Event<_>| {
+            let display = get_display();
+            match event {
+                Event::Add(output) => {
+                    debug!("Adding bar(s) for monitor {:?}", &output.name);
+                    create_bars_for_monitor(&app, &display, &output, config.clone(), &global_state)
+                        .unwrap();
+                }
+                // TODO: Implement
+                Event::Remove(_) => {}
+                Event::Replace(_, _) => {}
+            }
+            Continue(true)
+        });
+    }
     // Ignore CLI args
     // Some are provided by swaybar_config but not currently supported
     app.run_with_args(&Vec::<&str>::new());
 }
 
-/// Loads the Ironbar config and interface.
+/// Closes all current bars and entirely reloads Ironbar.
-pub fn load_interface(app: &Application, global_state: &Rc<RefCell<GlobalState>>) {
+/// This re-reads the config file.
-    let display = Display::default().map_or_else(
+pub async fn reload(app: &Application, global_state: &Rc<RefCell<GlobalState>>) -> Result<()> {
+    info!("Closing existing bars");
+    let windows = app.windows();
+    for window in windows {
+        window.close();
+    }
+
+    let config = load_config();
+
+    let wl = wayland::get_client();
+    let wl = lock!(wl);
+    let outputs = wl.get_outputs();
+
+    let display = get_display();
+    for output in outputs.iter() {
+        create_bars_for_monitor(app, &display, output, config.clone(), global_state)?;
+    }
+
+    Ok(())
+}
+fn create_bars_for_monitor(
+    app: &Application,
+    display: &Display,
+    output: &OutputInfo,
+    config: Config,
+    global_state: &Rc<RefCell<GlobalState>>,
+) -> Result<()> {
+    let Some(monitor_name) = &output.name else {
+        return Ok(());
+    };
+
+    let monitor = match get_monitor(&monitor_name, display) {
+        Ok(monitor) => monitor,
+        Err(err) => return Err(err),
+    };
+
+    let Some(monitor_config) = config.get_monitor_config(&monitor_name) else {
+        return Ok(());
+    };
+
+    match monitor_config {
+        MonitorConfig::Single(config) => {
+            create_bar(&app, &monitor, &monitor_name, config, global_state)
+        }
+        MonitorConfig::Multiple(configs) => configs
+            .into_iter()
+            .map(|config| create_bar(&app, &monitor, &monitor_name, config, global_state))
+            .collect(),
+    }
+}
+fn get_display() -> Display {
+    Display::default().map_or_else(
         || {
             let report = Report::msg("Failed to get default GTK display");
             error!("{:?}", report);
             exit(ExitCode::GtkDisplay as i32)
         },
         |display| display,
-    );
+    )
+}
 
-    let mut config = env::var("IRONBAR_CONFIG")
+fn get_monitor(name: &str, display: &Display) -> Result<Monitor> {
+    let wl = wayland::get_client();
+    let wl = lock!(wl);
+    let outputs = wl.get_outputs();
+
+    let monitor = (0..display.n_monitors()).into_iter().find_map(|i| {
+        let monitor = display.monitor(i)?;
+        let output = outputs.get(i as usize)?;
+
+        let is_match = output.name.as_ref().map(|n| n == name).unwrap_or_default();
+        if is_match {
+            Some(monitor)
+        } else {
+            None
+        }
+    });
+
+    monitor.ok_or_else(|| Report::msg(error::ERR_OUTPUTS))
+}
+
+fn load_config() -> Config {
+    let config = env::var("IRONBAR_CONFIG")
         .map_or_else(
             |_| ConfigLoader::new("ironbar").find_and_load(),
             ConfigLoader::load,
@@ -182,79 +292,7 @@ pub fn load_interface(app: &Application, global_state: &Rc<RefCell<GlobalState>>
         });
 
     debug!("Loaded config file");
-
+    config
-    #[cfg(feature = "ipc")]
-    if let Some(ironvars) = config.ironvar_defaults.take() {
-        let variable_manager = ironvar::get_variable_manager();
-        for (k, v) in ironvars {
-            if write_lock!(variable_manager).set(k.clone(), v).is_err() {
-                warn!("Ignoring invalid ironvar: '{k}'");
-            }
-        }
-    }
-
-    if let Err(err) = create_bars(app, &display, &config, global_state) {
-        error!("{:?}", err);
-        exit(ExitCode::CreateBars as i32);
-    }
-
-    debug!("Created bars");
-}
-
-/// Creates each of the bars across each of the (configured) outputs.
-fn create_bars(
-    app: &Application,
-    display: &Display,
-    config: &Config,
-    global_state: &Rc<RefCell<GlobalState>>,
-) -> Result<()> {
-    let wl = wayland::get_client();
-    let outputs = lock!(wl).get_outputs();
-
-    debug!("Received {} outputs from Wayland", outputs.len());
-    debug!("Outputs: {:?}", outputs);
-
-    let num_monitors = display.n_monitors();
-
-    for i in 0..num_monitors {
-        let monitor = display
-            .monitor(i)
-            .ok_or_else(|| Report::msg(error::ERR_OUTPUTS))?;
-        let output = outputs
-            .get(i as usize)
-            .ok_or_else(|| Report::msg(error::ERR_OUTPUTS))?;
-
-        let Some(monitor_name) = &output.name else {
-            continue;
-        };
-
-        config.monitors.as_ref().map_or_else(
-            || {
-                info!("Creating bar on '{}'", monitor_name);
-                create_bar(app, &monitor, monitor_name, config.clone(), global_state)
-            },
-            |config| {
-                let config = config.get(monitor_name);
-                match &config {
-                    Some(MonitorConfig::Single(config)) => {
-                        info!("Creating bar on '{}'", monitor_name);
-                        create_bar(app, &monitor, monitor_name, config.clone(), global_state)
-                    }
-                    Some(MonitorConfig::Multiple(configs)) => {
-                        for config in configs {
-                            info!("Creating bar on '{}'", monitor_name);
-                            create_bar(app, &monitor, monitor_name, config.clone(), global_state)?;
-                        }
-
-                        Ok(())
-                    }
-                    _ => Ok(()),
-                }
-            },
-        )?;
-    }
-
-    Ok(())
 }
 
 /// Blocks on a `Future` until it resolves.