`4f3be95`

feat(ipc): add event subscriptions for bar integration

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 3 months ago

SHA: 4f3be95ea06bc7b1cf5159fa895b7a4857c13420
Parents: 6b55ac2
Tree: 6943e61

2 changed files

Status	File	+	-
M	`garnotify/src/daemon.rs`	50	1
M	`garnotify/src/ipc.rs`	138	3

garnotify/src/daemon.rsmodified

  use crate::config::{self, Config};
  use crate::dbus::NotificationsService;
 -use crate::ipc::{Command, IpcRequest, IpcServer, Response};
 +use crate::ipc::{Command, Event, EventBroadcaster, IpcRequest, IpcServer, Response};
  use crate::notification::{
      new_shared_store, CloseReason, History, Notification, NotificationEvent,
      SharedNotificationStore, UrgencyTimeouts,
      config: Arc<Config>,
      ipc_server: IpcServer,
      ipc_rx: Receiver<IpcRequest>,
 +    /// Event broadcaster for subscribers
 +    broadcaster: EventBroadcaster,
      dbus_service: Option<NotificationsService>,
      notification_store: SharedNotificationStore,
      notification_event_rx: tokio::sync::mpsc::Receiver<NotificationEvent>,
      /// Create a new daemon
      pub fn new(config: Config) -> Result<Self> {
          let (ipc_server, ipc_rx) = IpcServer::new();
 +        let broadcaster = ipc_server.broadcaster();
          // Create notification event channel
          let (event_tx, event_rx) = tokio::sync::mpsc::channel(100);
              config,
              ipc_server,
              ipc_rx,
 +            broadcaster,
              dbus_service: None,
              notification_store,
              notification_event_rx: event_rx,
                      return;
+                 }
 +                // Broadcast event to subscribers
 +                self.broadcaster.broadcast(&Event::NotificationNew {
 +                    id: notification.id,
 +                    app_name: notification.app_name.clone(),
 +                    summary: notification.summary.clone(),
 +                    body: notification.body.clone(),
 +                    urgency: format!("{:?}", notification.hints.urgency),
 +                });
++
 +                // Broadcast count change
 +                let count = {
 +                    let store = self.notification_store.lock().await;
 +                    store.count()
 +                };
 +                self.broadcaster.broadcast(&Event::CountChanged { count });
++
                  // Check DND mode before showing popup
                  let should_show = if !self.paused {
                      true
                      "Notification updated: id={} summary=\"{}\"",
                      notification.id, notification.summary
                  );
++
 +                // Broadcast event to subscribers
 +                self.broadcaster.broadcast(&Event::NotificationUpdated {
 +                    id: notification.id,
 +                    summary: notification.summary.clone(),
 +                    body: notification.body.clone(),
 +                });
++
                  // Update popup
                  self.send_ui_command(PopupCommand::Update {
                      id: notification.id,
              NotificationEvent::Closed { id, reason } => {
                  info!("Notification {} closed: reason={:?}", id, reason);
 +                // Broadcast event to subscribers
 +                self.broadcaster.broadcast(&Event::NotificationClosed {
 +                    id,
 +                    reason: format!("{:?}", reason),
 +                });
++
                  // Close popup
                  self.close_notification_popup(id, reason.clone());
                      history.push(notification);
+                 }
 +                // Broadcast count change
 +                let count = {
 +                    let store = self.notification_store.lock().await;
 +                    store.count()
 +                };
 +                self.broadcaster.broadcast(&Event::CountChanged { count });
++
                  // Emit D-Bus signal
                  if let Some(ref dbus) = self.dbus_service {
                      if let Err(e) = dbus.emit_closed(id, reason).await {
                      if paused { "enabled" } else { "disabled" },
                      self.pause_level
                  );
++
 +                // Broadcast to subscribers
 +                self.broadcaster.broadcast(&Event::PausedChanged { paused });
++
                  Response::ok_with_message(format!(
                      "DND {}",
                      if paused { "enabled" } else { "disabled" }
                  ))
+             }
 +            Command::Subscribe => {
 +                // Subscribe is handled in the IPC layer, shouldn't reach here
 +                Response::ok_with_message("Subscribed")
 +            }
              Command::IsPaused => {
                  let data = serde_json::json!({
                      "paused": self.paused,

garnotify/src/ipc.rsmodified

  use anyhow::{Context, Result};
  use serde::{Deserialize, Serialize};
 -use std::io::{BufRead, BufReader, Write};
 +use std::io::{BufRead, BufReader, Read, Write};
  use std::os::unix::net::{UnixListener, UnixStream};
  use std::path::PathBuf;
  use std::sync::mpsc::{self, Receiver, Sender};
      Reload,
      /// Get daemon status
      Status,
 +    /// Subscribe to notification events (keeps connection open)
 +    Subscribe,
      /// Quit the daemon
      Quit,
+ }
 +/// Notification events sent to subscribers
 +#[derive(Debug, Clone, Serialize, Deserialize)]
 +#[serde(tag = "event", rename_all = "snake_case")]
 +pub enum Event {
 +    /// New notification created
 +    NotificationNew {
 +        id: u32,
 +        app_name: String,
 +        summary: String,
 +        body: String,
 +        urgency: String,
 +    },
 +    /// Notification closed
 +    NotificationClosed {
 +        id: u32,
 +        reason: String,
 +    },
 +    /// Notification updated/replaced
 +    NotificationUpdated {
 +        id: u32,
 +        summary: String,
 +        body: String,
 +    },
 +    /// DND state changed
 +    PausedChanged {
 +        paused: bool,
 +    },
 +    /// Count changed (for bar widgets)
 +    CountChanged {
 +        count: usize,
 +    },
 +}
++
  /// IPC response
  #[derive(Debug, Clone, Serialize, Deserialize)]
  pub struct Response {
      pub response_tx: std::sync::mpsc::Sender<Response>,
+ }
 +/// A subscriber connection for streaming events
 +pub struct Subscriber {
 +    stream: UnixStream,
 +}
++
 +impl Subscriber {
 +    /// Send an event to this subscriber
 +    pub fn send(&mut self, event: &Event) -> Result<()> {
 +        let json = serde_json::to_string(event)?;
 +        writeln!(self.stream, "{}", json)?;
 +        self.stream.flush()?;
 +        Ok(())
 +    }
 +}
++
  /// IPC server for the daemon
  pub struct IpcServer {
      socket_path: PathBuf,
      listener: Option<UnixListener>,
      tx: Sender<IpcRequest>,
 +    /// Active event subscribers
 +    subscribers: std::sync::Arc<std::sync::Mutex<Vec<Subscriber>>>,
+ }
  impl IpcServer {
              socket_path: socket_path(),
              listener: None,
              tx,
 +            subscribers: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
          };
          (server, rx)
+     }
 +    /// Get a handle for broadcasting events to subscribers
 +    pub fn broadcaster(&self) -> EventBroadcaster {
 +        EventBroadcaster {
 +            subscribers: self.subscribers.clone(),
 +        }
 +    }
++
      /// Start listening for connections
      pub fn start(&mut self) -> Result<()> {
          // Remove stale socket
          info!("IPC server listening on {}", self.socket_path.display());
          let tx = self.tx.clone();
 +        let subscribers = self.subscribers.clone();
          self.listener = Some(listener.try_clone()?);
          // Spawn listener thread
                  match stream {
                      Ok(stream) => {
                          let tx = tx.clone();
 +                        let subscribers = subscribers.clone();
                          thread::spawn(move || {
 -                            if let Err(e) = handle_client(stream, tx) {
 +                            if let Err(e) = handle_client(stream, tx, subscribers) {
                                  error!("Client error: {}", e);
+                             }
                          });
+     }
+ }
 +/// Handle for broadcasting events to all subscribers
 +#[derive(Clone)]
 +pub struct EventBroadcaster {
 +    subscribers: std::sync::Arc<std::sync::Mutex<Vec<Subscriber>>>,
 +}
++
 +impl EventBroadcaster {
 +    /// Broadcast an event to all subscribers, removing dead connections
 +    pub fn broadcast(&self, event: &Event) {
 +        let mut subs = match self.subscribers.lock() {
 +            Ok(s) => s,
 +            Err(_) => return,
 +        };
++
 +        // Send to all, track failures
 +        let mut failed = Vec::new();
 +        for (i, sub) in subs.iter_mut().enumerate() {
 +            if sub.send(event).is_err() {
 +                failed.push(i);
 +            }
 +        }
++
 +        // Remove failed subscribers (in reverse to preserve indices)
 +        for i in failed.into_iter().rev() {
 +            subs.remove(i);
 +            debug!("Removed dead subscriber");
 +        }
 +    }
++
 +    /// Get current subscriber count
 +    pub fn subscriber_count(&self) -> usize {
 +        self.subscribers.lock().map(|s| s.len()).unwrap_or(0)
 +    }
 +}
++
  /// Handle a client connection
 -fn handle_client(mut stream: UnixStream, tx: Sender<IpcRequest>) -> Result<()> {
 +fn handle_client(
 +    mut stream: UnixStream,
 +    tx: Sender<IpcRequest>,
 +    subscribers: std::sync::Arc<std::sync::Mutex<Vec<Subscriber>>>,
 +) -> Result<()> {
      let reader = BufReader::new(stream.try_clone()?);
      for line in reader.lines() {
          debug!("Received: {}", line);
          let response = match serde_json::from_str::<Command>(&line) {
 +            Ok(Command::Subscribe) => {
 +                // Send acknowledgment then add to subscribers
 +                let response = Response::ok_with_message("Subscribed to notification events");
 +                let response_json = serde_json::to_string(&response)?;
 +                writeln!(stream, "{}", response_json)?;
 +                stream.flush()?;
++
 +                // Add to subscribers list
 +                if let Ok(mut subs) = subscribers.lock() {
 +                    let sub_stream = stream.try_clone()?;
 +                    subs.push(Subscriber { stream: sub_stream });
 +                    info!("New subscriber connected, total: {}", subs.len());
 +                }
++
 +                // Keep connection open - block on read until client disconnects
 +                let mut buf = [0u8; 1];
 +                loop {
 +                    match stream.read(&mut buf) {
 +                        Ok(0) => {
 +                            debug!("Subscriber disconnected");
 +                            break;
 +                        }
 +                        Err(_) => {
 +                            debug!("Subscriber connection error");
 +                            break;
 +                        }
 +                        Ok(_) => {
 +                            // Ignore any data from subscriber (they should only read)
 +                        }
 +                    }
 +                }
++
 +                return Ok(());
 +            }
              Ok(cmd) => {
                  // Create response channel
                  let (response_tx, response_rx) = mpsc::channel();