feat(subscriptions): add and remove subscribers

This commit adds two new commands to add and remove subscribers to
WindowManagerEvent and SocketMessage notifications after they have been
handled by komorebi.

Interprocess communication is achieved using Named Pipes; the
subscribing process must first create the Named Pipe, and then run the
'add-subscriber' command, specifying the pipe name as the argument
(without the pipe filesystem path prepended).

Whenever a pipe is closing or has been closed, komorebi will flag this
as a stale subscription and remove it automatically.

resolve #54

Cargo.lock

@@ -525,6 +525,7 @@ dependencies = [
  "hotwatch",
  "komorebi-core",
  "lazy_static",
+ "miow 0.3.7",
  "nanoid",
  "parking_lot",
  "paste",
@@ -662,7 +663,7 @@ dependencies = [
  "kernel32-sys",
  "libc",
  "log",
- "miow",
+ "miow 0.2.2",
  "net2",
  "slab",
  "winapi 0.2.8",
@@ -692,6 +693,15 @@ dependencies = [
  "ws2_32-sys",
 ]
 
+[[package]]
+name = "miow"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
+dependencies = [
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "nanoid"
 version = "0.4.0"

README.md

@@ -276,6 +276,8 @@ start                         Start komorebi.exe as a background process
 stop                          Stop the komorebi.exe process and restore all hidden windows
 state                         Show a JSON representation of the current window manager state
 query                         Query the current window manager state
+add-subscriber                Subscribe to all komorebi events on a named pipe
+remove-subscriber             Subscribe to all komorebi events on a named pipe
 log                           Tail komorebi.exe's process logs (cancel with Ctrl-C)
 quick-save                    Quicksave the current resize layout dimensions
 quick-load                    Load the last quicksaved resize layout dimensions
@@ -393,6 +395,7 @@ used [is available here](komorebi.sample.with.lib.ahk).
 - [x] Helper library for AutoHotKey
 - [x] View window manager state
 - [x] Query window manager state
+- [x] Subscribe to event and message notifications
 
 ## Development
 
@@ -450,3 +453,34 @@ representation of the `State` struct, which includes the current state of `Windo
 
 This may also be polled to build further integrations and widgets on top of (if you ever wanted to build something
 like [Stackline](https://github.com/AdamWagner/stackline) for Windows, you could do it by polling this command).
+
+## Window Manager Event Subscriptions
+
+It is also possible to subscribe to notifications of every `WindowManagerEvent` and `SocketMessage` handled
+by `komorebi` using [Named Pipes](https://docs.microsoft.com/en-us/windows/win32/ipc/named-pipes).
+
+First, your application must create a named pipe. Once the named pipe has been created, run the following command:
+
+```powershell
+komorebic.exe add-subscriber <your pipe name>
+```
+
+Note that you do not have to incldue the full path of the named pipe, just the name. If the named pipe
+exists, `komorebi` will start pushing JSON data of successfully handled events and messages:
+
+```json lines
+{"type":"AddSubscriber","content":"test-pipe"}
+{"type":"FocusWindow","content":"Up"}
+{"type":"FocusChange","content":["SystemForeground",{"hwnd":1443930,"title":"komorebi – README.md","exe":"idea64.exe","class":"SunAwtFrame","rect":{"left":1539,"top":60,"right":1520,"bottom":821}}]}
+{"type":"MonitorPoll","content":["ObjectCreate",{"hwnd":2624200,"title":"OLEChannelWnd","exe":"explorer.exe","class":"OleMainThreadWndClass","rect":{"left":0,"top":0,"right":0,"bottom":0}}]}
+{"type":"FocusWindow","content":"Left"}
+{"type":"FocusChange","content":["SystemForeground",{"hwnd":2558668,"title":"Windows PowerShell","exe":"WindowsTerminal.exe","class":"CASCADIA_HOSTING_WINDOW_CLASS","rect":{"left":13,"top":60,"right":1520,"bottom":1655}}]}
+{"type":"FocusWindow","content":"Right"}
+{"type":"FocusChange","content":["SystemForeground",{"hwnd":1443930,"title":"komorebi – README.md","exe":"idea64.exe","class":"SunAwtFrame","rect":{"left":1539,"top":60,"right":1520,"bottom":821}}]}
+{"type":"FocusWindow","content":"Down"}
+{"type":"FocusChange","content":["SystemForeground",{"hwnd":67344,"title":"Windows PowerShell","exe":"WindowsTerminal.exe","class":"CASCADIA_HOSTING_WINDOW_CLASS","rect":{"left":1539,"top":894,"right":757,"bottom":821}}]}
+```
+
+You may then filter on the `type` key to listen to the events that you are interested in. For a full list of possible
+notification types, refer to the enum variants of `WindowManagerEvent` in `komorebi` and `SocketMessage`
+in `komorebi-core`.

komorebi-core/src/lib.rs

@@ -31,6 +31,7 @@ pub mod operation_direction;
 pub mod rect;
 
 #[derive(Clone, Debug, Serialize, Deserialize, Display)]
+#[serde(tag = "type", content = "content")]
 pub enum SocketMessage {
     // Window / Container Commands
     FocusWindow(OperationDirection),
@@ -92,16 +93,14 @@ pub enum SocketMessage {
     Query(StateQuery),
     FocusFollowsMouse(FocusFollowsMouseImplementation, bool),
     ToggleFocusFollowsMouse(FocusFollowsMouseImplementation),
+    AddSubscriber(String),
+    RemoveSubscriber(String),
 }
 
 impl SocketMessage {
     pub fn as_bytes(&self) -> Result<Vec<u8>> {
         Ok(serde_json::to_string(self)?.as_bytes().to_vec())
     }
-
-    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
-        Ok(serde_json::from_slice(bytes)?)
-    }
 }
 
 impl FromStr for SocketMessage {

komorebi/Cargo.toml

@@ -38,6 +38,7 @@ uds_windows = "1"
 which = "4"
 winput = "0.2"
 winvd = "0.0.20"
+miow = "0.3"
 
 [features]
 deadlock_detection = []

komorebi/src/main.rs

@@ -2,6 +2,8 @@
 #![allow(clippy::missing_errors_doc)]
 
 use std::collections::HashMap;
+use std::fs::File;
+use std::io::Write;
 use std::process::Command;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
@@ -79,6 +81,8 @@ lazy_static! {
         "mstsc.exe".to_string(),
         "vcxsrv.exe".to_string(),
     ]));
+    static ref SUBSCRIPTION_PIPES: Arc<Mutex<HashMap<String, File>>> =
+        Arc::new(Mutex::new(HashMap::new()));
 }
 
 pub static CUSTOM_FFM: AtomicBool = AtomicBool::new(false);
@@ -182,6 +186,40 @@ pub fn load_configuration() -> Result<()> {
     Ok(())
 }
 
+pub fn notify_subscribers(notification: &str) -> Result<()> {
+    let mut stale_subscriptions = vec![];
+    let mut subscriptions = SUBSCRIPTION_PIPES.lock();
+    for (subscriber, pipe) in subscriptions.iter_mut() {
+        match writeln!(pipe, "{}", notification) {
+            Ok(_) => {
+                tracing::debug!("pushed notification to subscriber: {}", subscriber);
+            }
+            Err(error) => {
+                // ERROR_FILE_NOT_FOUND
+                // 2 (0x2)
+                // The system cannot find the file specified.
+
+                // ERROR_NO_DATA
+                // 232 (0xE8)
+                // The pipe is being closed.
+
+                // Remove the subscription; the process will have to subscribe again
+                if let Some(2 | 232) = error.raw_os_error() {
+                    let subscriber_cl = subscriber.clone();
+                    stale_subscriptions.push(subscriber_cl);
+                }
+            }
+        }
+    }
+
+    for subscriber in stale_subscriptions {
+        tracing::warn!("removing stale subscription: {}", subscriber);
+        subscriptions.remove(&subscriber);
+    }
+
+    Ok(())
+}
+
 #[cfg(feature = "deadlock_detection")]
 #[tracing::instrument]
 fn detect_deadlocks() {

komorebi/src/process_command.rs

@@ -11,6 +11,7 @@ use std::thread;
 
 use color_eyre::eyre::anyhow;
 use color_eyre::Result;
+use miow::pipe::connect;
 use parking_lot::Mutex;
 use uds_windows::UnixStream;
 
@@ -19,6 +20,7 @@ use komorebi_core::Rect;
 use komorebi_core::SocketMessage;
 use komorebi_core::StateQuery;
 
+use crate::notify_subscribers;
 use crate::window_manager;
 use crate::window_manager::WindowManager;
 use crate::windows_api::WindowsApi;
@@ -26,6 +28,7 @@ use crate::BORDER_OVERFLOW_IDENTIFIERS;
 use crate::CUSTOM_FFM;
 use crate::FLOAT_IDENTIFIERS;
 use crate::MANAGE_IDENTIFIERS;
+use crate::SUBSCRIPTION_PIPES;
 use crate::TRAY_AND_MULTI_WINDOW_IDENTIFIERS;
 use crate::WORKSPACE_RULES;
 
@@ -435,6 +438,19 @@ impl WindowManager {
                 workspace.set_resize_dimensions(resize);
                 self.update_focused_workspace(false)?;
             }
+            SocketMessage::AddSubscriber(subscriber) => {
+                let mut pipes = SUBSCRIPTION_PIPES.lock();
+                let pipe_path = format!(r"\\.\pipe\{}", subscriber);
+                let pipe = connect(&pipe_path).map_err(|_| {
+                    anyhow!("the named pipe '{}' has not yet been created; please create it before running this command", pipe_path)
+                })?;
+
+                pipes.insert(subscriber, pipe);
+            }
+            SocketMessage::RemoveSubscriber(subscriber) => {
+                let mut pipes = SUBSCRIPTION_PIPES.lock();
+                pipes.remove(&subscriber);
+            }
         };
 
         tracing::info!("processed");
@@ -459,7 +475,8 @@ impl WindowManager {
                 };
             }
 
-            self.process_command(message)?;
+            self.process_command(message.clone())?;
+            notify_subscribers(&serde_json::to_string(&message)?)?;
         }
 
         Ok(())

komorebi/src/process_event.rs

@@ -11,6 +11,7 @@ use komorebi_core::OperationDirection;
 use komorebi_core::Rect;
 use komorebi_core::Sizing;
 
+use crate::notify_subscribers;
 use crate::window_manager::WindowManager;
 use crate::window_manager_event::WindowManagerEvent;
 use crate::windows_api::WindowsApi;
@@ -316,6 +317,7 @@ impl WindowManager {
             .open(hwnd_json)?;
 
         serde_json::to_writer_pretty(&file, &known_hwnds)?;
+        notify_subscribers(&serde_json::to_string(&event)?)?;
 
         tracing::info!("processed: {}", event.window().to_string());
         Ok(())

komorebi/src/window.rs

@@ -236,6 +236,7 @@ impl Window {
             return Ok(true);
         }
 
+        #[allow(clippy::question_mark)]
         if self.title().is_err() {
             return Ok(false);
         }

komorebi/src/window_manager_event.rs

@@ -1,11 +1,14 @@
 use std::fmt::Display;
 use std::fmt::Formatter;
 
+use serde::Serialize;
+
 use crate::window::Window;
 use crate::winevent::WinEvent;
 use crate::OBJECT_NAME_CHANGE_ON_LAUNCH;
 
-#[derive(Debug, Copy, Clone)]
+#[derive(Debug, Copy, Clone, Serialize)]
+#[serde(tag = "type", content = "content")]
 pub enum WindowManagerEvent {
     Destroy(WinEvent, Window),
     FocusChange(WinEvent, Window),

komorebi/src/winevent.rs

@@ -1,3 +1,4 @@
+use serde::Serialize;
 use strum::Display;
 
 use bindings::Windows::Win32::UI::WindowsAndMessaging::EVENT_AIA_END;
@@ -85,7 +86,7 @@ use bindings::Windows::Win32::UI::WindowsAndMessaging::EVENT_UIA_EVENTID_START;
 use bindings::Windows::Win32::UI::WindowsAndMessaging::EVENT_UIA_PROPID_END;
 use bindings::Windows::Win32::UI::WindowsAndMessaging::EVENT_UIA_PROPID_START;
 
-#[derive(Clone, Copy, PartialEq, Debug, Display)]
+#[derive(Clone, Copy, PartialEq, Debug, Serialize, Display)]
 #[repr(u32)]
 #[allow(dead_code)]
 pub enum WinEvent {

komorebic.lib.sample.ahk

@@ -16,6 +16,14 @@ Query(state_query) {
     Run, komorebic.exe query %state_query%, , Hide
 }
 
+AddSubscriber(named_pipe) {
+    Run, komorebic.exe add-subscriber %named_pipe%, , Hide
+}
+
+RemoveSubscriber(named_pipe) {
+    Run, komorebic.exe remove-subscriber %named_pipe%, , Hide
+}
+
 Log() {
     Run, komorebic.exe log, , Hide
 }

komorebic/src/main.rs

@@ -314,6 +314,18 @@ struct LoadCustomLayout {
     path: String,
 }
 
+#[derive(Clap, AhkFunction)]
+struct AddSubscriber {
+    /// Name of the pipe to send notifications to (without "\\.\pipe\" prepended)
+    named_pipe: String,
+}
+
+#[derive(Clap, AhkFunction)]
+struct RemoveSubscriber {
+    /// Name of the pipe to stop sending notifications to (without "\\.\pipe\" prepended)
+    named_pipe: String,
+}
+
 #[derive(Clap)]
 #[clap(author, about, version, setting = AppSettings::DeriveDisplayOrder)]
 struct Opts {
@@ -332,6 +344,12 @@ enum SubCommand {
     /// Query the current window manager state
     #[clap(setting = AppSettings::ArgRequiredElseHelp)]
     Query(Query),
+    /// Subscribe to all komorebi events on a named pipe
+    #[clap(setting = AppSettings::ArgRequiredElseHelp)]
+    AddSubscriber(AddSubscriber),
+    /// Subscribe to all komorebi events on a named pipe
+    #[clap(setting = AppSettings::ArgRequiredElseHelp)]
+    RemoveSubscriber(RemoveSubscriber),
     /// Tail komorebi.exe's process logs (cancel with Ctrl-C)
     Log,
     /// Quicksave the current resize layout dimensions
@@ -895,6 +913,12 @@ fn main() -> Result<()> {
         SubCommand::Load(arg) => {
             send_message(&*SocketMessage::Load(resolve_windows_path(&arg.path)?).as_bytes()?)?;
         }
+        SubCommand::AddSubscriber(arg) => {
+            send_message(&*SocketMessage::AddSubscriber(arg.named_pipe).as_bytes()?)?;
+        }
+        SubCommand::RemoveSubscriber(arg) => {
+            send_message(&*SocketMessage::RemoveSubscriber(arg.named_pipe).as_bytes()?)?;
+        }
     }
 
     Ok(())