From e2d757bbb0e85af833a07fe9a4b52b2a189f84d3 Mon Sep 17 00:00:00 2001 From: LGUG2Z Date: Tue, 27 Oct 2026 14:41:18 -0700 Subject: [PATCH] feat(wm): subscribe to notifications A program such as a status bar has to follow the window manager, and so far it could only ask: komorebi on Windows and on macOS sends notifications for that. The daemon now keeps subscribers, each a Unix socket in its data directory that a program listens on, and after each event and each message it handles, it connects to every subscriber and writes the event or the message with the state after it. A subscriber can ask to be sent only what changed the state; a window that is shown and a new subscription are always sent, so a subscriber has the state as soon as it subscribes. A subscriber that cannot be connected to is dropped with its socket file and not tried again, and a stopping daemon closes an empty connection to each one. komorebic gains subscribe-socket and unsubscribe-socket, and komorebi-client subscribe and subscribe_with_options, which bind the socket and subscribe. The tests subscribe through real sockets against the fake desktop: a shown window is sent with the state that has it, a subscriber that filters is not sent a question that changed nothing, and one that is gone is dropped. --- README.md | 13 +- docs/cli/subscribe-socket.md | 14 + docs/cli/unsubscribe-socket.md | 14 + docs/dev/ARCHITECTURE.md | 66 ++++- docs/dev/DESIGN.md | 11 + docs/dev/PLATFORM.md | 9 +- docs/index.md | 6 +- docs/usage/running-komorebi.md | 3 + docs/usage/subscribing-to-notifications.md | 83 ++++++ komorebi-client/src/lib.rs | 160 ++++++++++++ komorebi-client/tests/subscribe.rs | 281 +++++++++++++++++++++ komorebi/src/core/mod.rs | 44 +++- komorebi/src/lib.rs | 196 +++++++++++++- komorebi/src/main.rs | 12 + komorebi/src/process_command.rs | 51 +++- komorebi/src/process_event.rs | 23 ++ komorebi/src/state.rs | 31 ++- komorebi/src/window_manager.rs | 25 +- komorebic/src/main.rs | 36 +++ komorebic/tests/cli.rs | 25 ++ 20 files changed, 1078 insertions(+), 25 deletions(-) create mode 100644 docs/cli/subscribe-socket.md create mode 100644 docs/cli/unsubscribe-socket.md create mode 100644 docs/usage/subscribing-to-notifications.md create mode 100644 komorebi-client/tests/subscribe.rs diff --git a/README.md b/README.md index 75687334..abb6b4ee 100644 --- a/README.md +++ b/README.md @@ -62,10 +62,12 @@ without a checkout of the toolchain. (`window_manager_event`, `window_manager_event_listener`, `process_event`), the `Ring` that holds monitors, workspaces and containers, and the lists in which some elements are locked in place. It answers the messages on the daemon's - socket (`core::SocketMessage`, `process_command`) and shows its state - (`state`). + socket (`core::SocketMessage`, `process_command`), shows its state (`state`), + and sends its subscribers a notification of each event and message it handles + (`notify_subscribers`). - [`komorebi-client`](komorebi-client): a library that sends messages to the - daemon and reads its answers, and re-exports the types they are made of. + daemon, reads its answers, subscribes to its notifications, and re-exports + the types they are made of. - [`komorebic`](komorebic): the command line. A subcommand that asks something of the daemon is one message on its socket; `start`, `log`, `data-directory`, `completions` and the hidden `docgen` work without the daemon. @@ -118,6 +120,11 @@ Everything else reaches the daemon as a message on its socket: one JSON `SocketMessage` per line, handled one at a time with the window manager locked. A message that asks a question is answered on the same connection. +A program such as a status bar can subscribe to the daemon's notifications: it +listens on a socket of its own, and after each event and each message, the +daemon connects to it and writes the event or the message with the state of +the window manager. + ## Documentation The user guide starts at [`docs/index.md`](docs/index.md), and diff --git a/docs/cli/subscribe-socket.md b/docs/cli/subscribe-socket.md new file mode 100644 index 00000000..647ec0aa --- /dev/null +++ b/docs/cli/subscribe-socket.md @@ -0,0 +1,14 @@ +```text +Subscribe to komorebi events using a Unix Domain Socket + +Usage: subscribe-socket + +Arguments: + + Name of the socket to send event notifications to + +Options: + -h, --help + Print help + +``` diff --git a/docs/cli/unsubscribe-socket.md b/docs/cli/unsubscribe-socket.md new file mode 100644 index 00000000..faaaf985 --- /dev/null +++ b/docs/cli/unsubscribe-socket.md @@ -0,0 +1,14 @@ +```text +Unsubscribe from komorebi events + +Usage: unsubscribe-socket + +Arguments: + + Name of the socket to stop sending event notifications to + +Options: + -h, --help + Print help + +``` diff --git a/docs/dev/ARCHITECTURE.md b/docs/dev/ARCHITECTURE.md index 28aa7885..9cb61ff1 100644 --- a/docs/dev/ARCHITECTURE.md +++ b/docs/dev/ARCHITECTURE.md @@ -7,8 +7,9 @@ tested, without one. This page describes that trait, who implements and calls it, where each of its rules is enforced, the model of the desktop the window manager tiles, the backend that reaches KWin, -how an event about a window becomes a tile, and how a message on the daemon's socket is -answered. It grows with the code: each part of the daemon that is added has its place here. +how an event about a window becomes a tile, how a message on the daemon's socket is +answered, and how subscribers are notified of both. It grows with the code: each part of the +daemon that is added has its place here. ## Vocabulary @@ -28,6 +29,10 @@ answered. It grows with the code: each part of the daemon that is added has its (retile, stop) or a question (the state, the monitors, one [`StateQuery`]). - **State**: the window manager's monitors, workspaces, containers and windows, and its settings, as the daemon answers a `State` message with them ([`State`]). +- **Subscriber**: a program that listens on a socket of its own in [`DATA_DIR`] and has + asked the daemon, by the name of that socket, to send it notifications. +- **Notification**: what a subscriber is sent after each event and each message the window + manager handles: the event or the message, and the state after it ([`Notification`]). - **Compositor**: the program that owns the screen and the windows. On KDE Plasma it is KWin. - **Backend**: the implementation of the [`Backend`](../../komorebi/src/backend/mod.rs) @@ -191,6 +196,37 @@ connection close with no answer. what it set up in the compositor (the KWin backend unloads the bridge), and exits with status 0. `SIGINT` and `SIGTERM` do the same from `main.rs`, with the exit status 130. +## Notifications + +A subscriber binds a socket in [`DATA_DIR`] and sends `AddSubscriberSocket` with its name, +or `AddSubscriberSocketWithOptions` with its name and [`SubscribeOptions`] +([`subscribe`], [`subscribe_with_options`], `komorebic subscribe-socket`). +[`WindowManager::process_command`] records it in [`SUBSCRIPTION_SOCKETS`], and its options in +[`SUBSCRIPTION_SOCKET_OPTIONS`]. `RemoveSubscriberSocket` takes it out again. The +subscribers are kept in memory only, and a daemon that restarts has none. + +At the end of [`WindowManager::process_event`] and of [`WindowManager::process_command`], +the window manager calls [`notify_subscribers`] with a [`Notification`]: the event or the +message ([`NotificationEvent`]), and the [`State`] after it. A [`State`] made before the event +or the message is compared with the window manager after it ([`State::has_been_modified`]); +the monitors are compared down to the windows, and a window by its id only. An event about a +window komorebi does not manage, and a message that fails, return before this and are not +notified. + +[`notify_subscribers`] serializes the notification once, and for each subscriber connects to +its socket, writes the JSON and closes the connection, all on the thread that handled the +event or the message, with the window manager still locked. A subscriber with +`filter_state_changes` is skipped when the state did not change, unless the notification is +about a window being shown or a subscriber being added. A subscriber that cannot be +connected to is removed, with its socket file, after the loop. Writing to a subscriber that +was connected to has no timeout, and a failed write returns an error from the call that +handled the event or the message. A subscriber that stops reading therefore holds the window +manager, and with it the thread that handles events: the backend's events wait in the +channel of [`window_manager_event_listener`], and past 20 the backend drops them. + +When the daemon stops ([`WindowManager::stop`], or `SIGINT` and `SIGTERM` in `main.rs`), it +connects to each subscriber and shuts the connection down without writing anything. + ## The parts - [`backend`](../../komorebi/src/backend/mod.rs): the trait, the two data types, and the @@ -220,9 +256,13 @@ status 0. `SIGINT` and `SIGTERM` do the same from `main.rs`, with the exit statu [`bridge.js`](../../komorebi/src/kwin/bridge.js), the bridge. - [`process_command`](../../komorebi/src/process_command.rs): the thread that listens on the socket, and what the window manager does with each message. -- [`state`](../../komorebi/src/state.rs): the [`State`] a `State` message is answered with. +- [`state`](../../komorebi/src/state.rs): the [`State`] a `State` message is answered with, + and whether an event or a message changed it. +- [`lib.rs`](../../komorebi/src/lib.rs): the subscribers, the [`Notification`], and + [`notify_subscribers`]. - [`komorebi-client`](../../komorebi-client/src/lib.rs): a library that writes messages on - the socket and reads the answers, for programs that talk to the daemon. + the socket and reads the answers, and subscribes to notifications, for programs that talk + to the daemon. - [`komorebic`](../../komorebic/src/main.rs): the command line. A subcommand that asks something of the daemon is one message, sent with `komorebi-client`; `start` runs the daemon, `log` follows its log file in [`LOG_DIR`], and the hidden `docgen` writes @@ -282,6 +322,7 @@ the fake. The window manager is its only caller, through [`backend`]. The contra | Messages are handled one at a time, and in order on one connection | framework path: [`read_commands_uds`] reads a connection line by line and handles each message with the window manager locked | `the_messages_on_one_connection_are_handled_in_order` | none | | A socket left behind by a daemon that did not stop cleanly does not keep the next one from starting | framework path: [`WindowManager::new`] removes the file before it binds | `a_socket_file_left_behind_is_replaced` | none | | A daemon that is told to stop leaves no socket behind | framework path: [`WindowManager::stop`] removes it before it exits | `the_daemon_answers_on_its_socket_and_stops_when_it_is_told_to` | `SIGKILL` | +| A subscriber that is gone is dropped, not retried | test | `a_subscriber_that_is_gone_is_dropped_and_not_tried_again` | none | | The bridge holds no window manager logic | convention, with every decision about an event in the table, which the `kwin::state` unit tests pin | the unit tests in `komorebi/src/kwin/state.rs` | none | | What the bridge sends is what the table reads | type checker: the JSDoc types of the messages in [`bridge.js`]; tests: the events of [`fixtures`] are what the bridge's tests check it sends and what the table's tests feed it | `what the bridge sends for a session is what the daemon's tests read`, `the_bridges_events_for_a_session_are_the_window_events_komorebi_expects` | none | | What the backend sends is what the bridge carries out | tests: the commands of [`fixtures`] are what the backend's tests check it sends and what the bridge's tests answer it with, checking what KWin is asked to do for each | `the_commands_of_a_session_go_to_the_bridge_as_the_bridges_tests_answer_it`, `what the bridge sends for a session is what the daemon's tests read`, `a command for a window KWin no longer has is given back, and the rest are carried out` | none | @@ -326,6 +367,9 @@ A window is shown, and ends up on a tile: `focus`, and the pointer follows it (on KWin, through its "Move Mouse to Focus" shortcut). 8. The index is rebuilt again, so the next event finds the window. +9. [`notify_subscribers`] sends every subscriber a [`Notification`] of the `Show`, with the + state that now has the window. A `Show` is sent to a subscriber with + `filter_state_changes` too, whether or not it changed the state. ## One message, end to end @@ -341,7 +385,8 @@ A window is shown, and ends up on a tile: every monitor with its workspaces, containers and windows. Each window is serialized with what the backend reports about it now: its frame, title, executable, class and path. 5. The state is written to the connection as pretty-printed JSON, and the index of managed - windows is rebuilt, as after every message. + windows is rebuilt, as after every message. The subscribers are sent a notification of + the `State` message, except those with `filter_state_changes`: it changed nothing. 6. There are no more lines, so `read_commands_uds` returns and the connection is closed. `send_query` has read to the end, and returns the JSON, which `komorebic` prints. @@ -362,6 +407,8 @@ A window is shown, and ends up on a tile: each message does, and the tests that send messages through a real socket. - [`komorebi-client/src/lib.rs`](../../komorebi-client/src/lib.rs): the client side of the socket. +- [`komorebi-client/tests/subscribe.rs`](../../komorebi-client/tests/subscribe.rs): + subscribers and the notifications they are sent, through real sockets. - [`komorebic/src/main.rs`](../../komorebic/src/main.rs): the subcommands, and [`komorebic/tests/cli.rs`](../../komorebic/tests/cli.rs), what each writes on the socket. - [`komorebi/src/kwin/mod.rs`](../../komorebi/src/kwin/mod.rs): the KWin backend, its D-Bus @@ -421,6 +468,13 @@ A window is shown, and ends up on a tile: [`SocketMessage`]: ../../komorebi/src/core/mod.rs [`StateQuery`]: ../../komorebi/src/core/mod.rs [`State`]: ../../komorebi/src/state.rs +[`State::has_been_modified`]: ../../komorebi/src/state.rs +[`SubscribeOptions`]: ../../komorebi/src/core/mod.rs +[`Notification`]: ../../komorebi/src/lib.rs +[`NotificationEvent`]: ../../komorebi/src/lib.rs +[`notify_subscribers`]: ../../komorebi/src/lib.rs +[`SUBSCRIPTION_SOCKETS`]: ../../komorebi/src/lib.rs +[`SUBSCRIPTION_SOCKET_OPTIONS`]: ../../komorebi/src/lib.rs [`WindowManager::new`]: ../../komorebi/src/window_manager.rs [`WindowManager::stop`]: ../../komorebi/src/window_manager.rs [`WindowManager::process_command`]: ../../komorebi/src/process_command.rs @@ -429,3 +483,5 @@ A window is shown, and ends up on a tile: [`send_query`]: ../../komorebi-client/src/lib.rs [`send_message`]: ../../komorebi-client/src/lib.rs [`send_batch`]: ../../komorebi-client/src/lib.rs +[`subscribe`]: ../../komorebi-client/src/lib.rs +[`subscribe_with_options`]: ../../komorebi-client/src/lib.rs diff --git a/docs/dev/DESIGN.md b/docs/dev/DESIGN.md index 67ab1d07..1454bb66 100644 --- a/docs/dev/DESIGN.md +++ b/docs/dev/DESIGN.md @@ -171,3 +171,14 @@ Because the daemon takes no input but its socket. `komorebic` turns a command li message on the socket and, for a question, prints the answer. A key binding, a script and a person at a terminal all drive komorebi the same way, by running `komorebic` (the design notes of komorebi for Mac, linked above). + +## Why does komorebi send notifications to subscribers? + +So that a program such as a status bar can follow the window manager. The README of komorebi +for Windows demonstrates a custom status bar integrated through its notifications +([Window Manager Event Subscriptions](https://github.com/LGUG2Z/komorebi#window-manager-event-subscriptions)), +and komorebi for Mac sends the same notifications to Unix sockets, which is also how +`komorebic` talks to its daemon +([its README](https://github.com/LGUG2Z/komorebi-for-mac#window-manager-event-subscriptions)). +A subscriber here is a Unix socket, as on macOS. The named pipes that komorebi on Windows +sends notifications to are not ported, since the socket subscribers do the same. diff --git a/docs/dev/PLATFORM.md b/docs/dev/PLATFORM.md index e1870181..d56e38fe 100644 --- a/docs/dev/PLATFORM.md +++ b/docs/dev/PLATFORM.md @@ -89,10 +89,10 @@ an owner again; the VM test `vm-kwin-restart` restarts KWin twice under a runnin #### What komorebi leaves behind When it is stopped with `SIGINT`, `SIGTERM` or a `Stop` message on its socket, it asks KWin -to unload the script and removes the socket. The file `bridge.js` stays in the runtime -directory, which is removed at logout. A daemon that is killed leaves the script running in -KWin, reporting to nobody, and the socket file in the runtime directory, until the next start -replaces both or the session ends. +to unload the script, closes an empty connection to each subscriber, and removes the socket. +The file `bridge.js` stays in the runtime directory, which is removed at logout. A daemon +that is killed leaves the script running in KWin, reporting to nobody, and the socket file in +the runtime directory, until the next start replaces both or the session ends. ### A headless KWin @@ -144,6 +144,7 @@ What komorebi calls: |---|---|---| | `$XDG_RUNTIME_DIR/komorebi/bridge.js` | the script KWin loads, with its configuration (the D-Bus names and the hold time of `Poll`) on its first line | written at every start; the directory is removed at logout | | `$XDG_RUNTIME_DIR/komorebi/komorebi.sock` | the command socket | bound once the daemon is connected to KWin, after a file left there is removed; removed when the daemon stops | +| `$XDG_RUNTIME_DIR/komorebi/NAME` | the socket of a subscriber, which the subscriber binds and names in its subscription | removed by the daemon, with the subscriber, when it cannot connect to it | | `$XDG_DATA_HOME/komorebi/komorebi.log.YYYY-MM-DD` | the daemon's log, with colors, which `komorebic log` follows | a file a day (the UTC date), never removed by komorebi | | `$XDG_DATA_HOME/komorebi/komorebi_plaintext.log.YYYY-MM-DD` | the same log without colors | as above | diff --git a/docs/index.md b/docs/index.md index ef04d17f..7a5f98a1 100644 --- a/docs/index.md +++ b/docs/index.md @@ -29,8 +29,10 @@ key bindings --------> komorebic <------> komorebi ## Getting started [Running komorebi](usage/running-komorebi.md) explains how to get the two programs, start and -stop the daemon, ask it what it is doing and read its log. The [CLI reference](cli/start.md) -has a page for every `komorebic` command, generated from its help. +stop the daemon, ask it what it is doing and read its log. +[Subscribing to notifications](usage/subscribing-to-notifications.md) explains how a program +such as a status bar follows what komorebi does. The [CLI reference](cli/start.md) has a page +for every `komorebic` command, generated from its help. ## Community diff --git a/docs/usage/running-komorebi.md b/docs/usage/running-komorebi.md index 748254bd..0e0eb08f 100644 --- a/docs/usage/running-komorebi.md +++ b/docs/usage/running-komorebi.md @@ -66,6 +66,9 @@ a window that was moved by hand back on its tile. When the daemon is not running, these commands fail: there is no socket to write to. +A program that wants to know what komorebi does without asking can subscribe to its +notifications instead ([Subscribing to notifications](subscribing-to-notifications.md)). + ## The socket and the data directory The daemon listens on `komorebi.sock` in its data directory, `$XDG_RUNTIME_DIR/komorebi` diff --git a/docs/usage/subscribing-to-notifications.md b/docs/usage/subscribing-to-notifications.md new file mode 100644 index 00000000..79f34b8c --- /dev/null +++ b/docs/usage/subscribing-to-notifications.md @@ -0,0 +1,83 @@ +# Subscribing to notifications + +A program such as a status bar, a widget or a script can follow what komorebi does by +subscribing to its notifications. After each event the window manager handles (a window +was shown, closed, focused, minimized or restored) and each message it takes on its socket, +the daemon sends every subscriber a notification: the event or the message, and the state +of the window manager after it. + +## Subscribing + +A subscriber listens on a Unix socket of its own in the daemon's data directory, +`$XDG_RUNTIME_DIR/komorebi` (on NixOS and most systemd distributions, +`/run/user//komorebi`), and tells the daemon the name of the socket with +[`komorebic subscribe-socket`](../cli/subscribe-socket.md). The socket has to exist before +it subscribes: the daemon sends the notification of the subscription at once, and removes a +subscriber it cannot connect to. + +With [socat](http://www.dest-unreach.org/socat/), a shell can subscribe and print every +notification: + +```sh +socat -u UNIX-LISTEN:"$XDG_RUNTIME_DIR/komorebi/bar.sock",fork STDOUT & +komorebic subscribe-socket bar.sock +``` + +[`komorebic unsubscribe-socket bar.sock`](../cli/unsubscribe-socket.md) stops the +notifications. + +## Notifications + +The daemon sends each notification on a new connection to the subscriber's socket: it +connects, writes one JSON object, and closes the connection. There is no newline after the +object. The first notification a subscriber receives is the one of its subscription. + +A notification has two fields: + +- `event`: the event or the message. An event is a window event, with the signal it came + from and the window, such as + `{"type":"Show","content":[{"type":"KWin","content":"WindowAdded"},{"id":...}]}`. A + message is written the way a client writes it on the daemon's socket, such as + `{"type":"Retile"}`. +- `state`: the state of the window manager after it, with the same fields as what + [`komorebic state`](../cli/state.md) prints, on one line. + +An event about a window that komorebi does not manage, and a message that fails, are not +sent. + +## Only when something changed + +A subscriber can ask to be sent a notification only when the event or the message changed +the state, with the `filter_state_changes` option. It is still sent the notification of a +window that is shown and of every new subscription, so it receives the state as soon as it +subscribes. `komorebic subscribe-socket` has no option for it; a program writes the message +on the daemon's socket itself: + +```sh +printf '%s' '{"type":"AddSubscriberSocketWithOptions","content":["bar.sock",{"filter_state_changes":true}]}' | + socat - UNIX-CONNECT:"$XDG_RUNTIME_DIR/komorebi/komorebi.sock" +``` + +## When a subscriber goes away + +A subscriber that the daemon cannot connect to, because its program has exited or its +socket was removed, is removed at the next notification, and its socket file with it. The +daemon does not try it again: a program that wants notifications again subscribes again. +The daemon keeps its subscribers only while it runs, so after a restart every program has +to subscribe again. + +When the daemon stops, it connects to each subscriber and closes the connection without +writing anything. + +A subscriber should accept and read every connection as it comes. The daemon writes each +notification while the window manager is locked, and waits when the socket's buffer is +full, so a subscriber that stops reading can hold up the window manager. While it is held +up, the window events KWin reports wait for it, and once 20 are waiting the daemon drops +the ones that come after them. + +## In Rust + +The `komorebi-client` crate has `subscribe` and `subscribe_with_options`, which listen on +the socket, subscribe, and return the listener: each connection on it is one +`Notification`. `cargo doc --package komorebi-client --open` shows their documentation, +with an example. diff --git a/komorebi-client/src/lib.rs b/komorebi-client/src/lib.rs index c249345e..f06c7f7c 100644 --- a/komorebi-client/src/lib.rs +++ b/komorebi-client/src/lib.rs @@ -9,6 +9,10 @@ //! daemon that is not running is an [`std::io::ErrorKind::NotFound`] or a //! [`std::io::ErrorKind::ConnectionRefused`]. //! +//! [`subscribe`] and [`subscribe_with_options`] make the program a subscriber: they listen on a +//! socket of its own in [`struct@DATA_DIR`], and ask the daemon to send a [`Notification`] to it +//! after each event and each message the window manager handles. +//! //! The crate also re-exports the types the messages and the answers are made of, so that a client //! needs nothing else. //! @@ -51,6 +55,8 @@ pub use komorebi::DATA_DIR; pub use komorebi::LOG_DIR; +pub use komorebi::Notification; +pub use komorebi::NotificationEvent; pub use komorebi::container::Container; pub use komorebi::core::ApplicationIdentifier; pub use komorebi::core::Axis; @@ -66,6 +72,7 @@ pub use komorebi::core::Rect; pub use komorebi::core::Sizing; pub use komorebi::core::SocketMessage; pub use komorebi::core::StateQuery; +pub use komorebi::core::SubscribeOptions; pub use komorebi::core::WindowKind; pub use komorebi::core::pathext::PathExt; pub use komorebi::core::pathext::replace_env_in_path; @@ -77,6 +84,7 @@ use std::io::BufReader; use std::io::Read; use std::io::Write; use std::net::Shutdown; +use std::os::unix::net::UnixListener; use std::os::unix::net::UnixStream; use std::time::Duration; @@ -219,3 +227,155 @@ pub fn send_query(message: &SocketMessage) -> std::io::Result { Ok(response) } + +/// Listens on the socket `name` in [`struct@DATA_DIR`], and asks the daemon to send a +/// [`Notification`] to it after each event and each message the window manager handles +/// +/// A socket file that is left over at that path is removed first. Each notification comes on a +/// new connection, as JSON, and ends when the daemon closes the connection; the first is the +/// notification of the subscription itself. When the daemon stops, it connects and closes the +/// connection without writing anything. +/// +/// A subscriber that the daemon cannot connect to, because its program has exited or its socket +/// was removed, is removed at the next notification, with its socket file. It has to subscribe +/// again: the daemon does not try it again, and does not keep its subscribers when it restarts. +/// +/// # Errors +/// +/// - The socket file left over at that path cannot be removed. +/// - The socket cannot be bound, for example because [`struct@DATA_DIR`] does not exist. +/// - The subscription cannot be sent, for one of the reasons [`send_message`] gives. +/// +/// # Example +/// +/// ```rust +/// # use std::io::{BufRead, BufReader, Write}; +/// # let runtime = format!("/tmp/komorebi-client-doc-subscribe-{}", std::process::id()); +/// # std::fs::create_dir_all(format!("{runtime}/komorebi"))?; +/// # unsafe { std::env::set_var("XDG_RUNTIME_DIR", &runtime) }; // the sockets are in it +/// # let socket = komorebi_client::DATA_DIR.join("komorebi.sock"); +/// # let listener = std::os::unix::net::UnixListener::bind(&socket)?; +/// # let daemon = std::thread::spawn(move || -> std::io::Result { +/// # let (stream, _) = listener.accept()?; +/// # let mut message = String::new(); +/// # BufReader::new(stream).read_line(&mut message)?; +/// # let subscriber = komorebi_client::DATA_DIR.join("bar.sock"); +/// # let state = r#"{"monitors":{"elements":[],"focused":0},"work_area_offset":null,"mouse_follows_focus":true}"#; +/// # let notification = format!(r#"{{"event":{message},"state":{state}}}"#); +/// # std::os::unix::net::UnixStream::connect(subscriber)?.write_all(notification.as_bytes())?; +/// # Ok(message) +/// # }); +/// use std::io::Read; +/// +/// use komorebi_client::Notification; +/// use komorebi_client::NotificationEvent; +/// use komorebi_client::SocketMessage; +/// use komorebi_client::subscribe; +/// +/// let listener = subscribe("bar.sock")?; +/// +/// // the first notification is the one of the subscription +/// let (mut connection, _) = listener.accept()?; +/// let mut json = String::new(); +/// connection.read_to_string(&mut json)?; +/// let notification: Notification = serde_json::from_str(&json)?; +/// assert!(matches!( +/// notification.event, +/// NotificationEvent::Socket(SocketMessage::AddSubscriberSocket(name)) if name == "bar.sock" +/// )); +/// +/// # assert_eq!( +/// # daemon.join().unwrap()?, +/// # r#"{"type":"AddSubscriberSocket","content":"bar.sock"}"# +/// # ); +/// # std::fs::remove_dir_all(&runtime)?; +/// # Ok::<(), Box>(()) +/// ``` +pub fn subscribe(name: &str) -> std::io::Result { + let socket = DATA_DIR.join(name); + + match std::fs::remove_file(&socket) { + Ok(()) => {} + Err(error) => match error.kind() { + std::io::ErrorKind::NotFound => {} + _ => { + return Err(error); + } + }, + }; + + let listener = UnixListener::bind(&socket)?; + + send_message(&SocketMessage::AddSubscriberSocket(name.to_string()))?; + + Ok(listener) +} + +/// The same as [`subscribe`], with the [`SubscribeOptions`] the daemon applies to this subscriber +/// +/// With `filter_state_changes`, the daemon sends a notification only when the event or the message +/// changed the [`State`], and always when a window is shown and when a subscriber is added. +/// +/// # Errors +/// +/// The same as [`subscribe`]. +/// +/// # Example +/// +/// ```rust +/// # use std::io::{BufRead, BufReader}; +/// # let runtime = format!("/tmp/komorebi-client-doc-options-{}", std::process::id()); +/// # std::fs::create_dir_all(format!("{runtime}/komorebi"))?; +/// # unsafe { std::env::set_var("XDG_RUNTIME_DIR", &runtime) }; // the sockets are in it +/// # let socket = komorebi_client::DATA_DIR.join("komorebi.sock"); +/// # let listener = std::os::unix::net::UnixListener::bind(&socket)?; +/// # let daemon = std::thread::spawn(move || -> std::io::Result { +/// # let (stream, _) = listener.accept()?; +/// # let mut message = String::new(); +/// # BufReader::new(stream).read_line(&mut message)?; +/// # Ok(message) +/// # }); +/// use komorebi_client::DATA_DIR; +/// use komorebi_client::SubscribeOptions; +/// use komorebi_client::subscribe_with_options; +/// +/// let options = SubscribeOptions { +/// filter_state_changes: true, +/// }; +/// let listener = subscribe_with_options("bar.sock", options)?; +/// +/// let socket = DATA_DIR.join("bar.sock"); +/// assert_eq!(listener.local_addr()?.as_pathname(), Some(socket.as_path())); +/// +/// # assert_eq!( +/// # daemon.join().unwrap()?, +/// # r#"{"type":"AddSubscriberSocketWithOptions","content":["bar.sock",{"filter_state_changes":true}]}"# +/// # ); +/// # std::fs::remove_dir_all(&runtime)?; +/// # Ok::<(), Box>(()) +/// ``` +pub fn subscribe_with_options( + name: &str, + options: SubscribeOptions, +) -> std::io::Result { + let socket = DATA_DIR.join(name); + + match std::fs::remove_file(&socket) { + Ok(()) => {} + Err(error) => match error.kind() { + std::io::ErrorKind::NotFound => {} + _ => { + return Err(error); + } + }, + }; + + let listener = UnixListener::bind(&socket)?; + + send_message(&SocketMessage::AddSubscriberSocketWithOptions( + name.to_string(), + options, + ))?; + + Ok(listener) +} diff --git a/komorebi-client/tests/subscribe.rs b/komorebi-client/tests/subscribe.rs new file mode 100644 index 00000000..5c22ee40 --- /dev/null +++ b/komorebi-client/tests/subscribe.rs @@ -0,0 +1,281 @@ +//! Subscribers against a window manager that listens the way the daemon does, with +//! [`listen_for_commands`], over an in-memory desktop with one monitor: each subscriber listens on a +//! socket of its own in a runtime directory of this process, subscribes with `komorebi-client`, and +//! reads the notifications the window manager sends it. +//! +//! The tests share the window manager, and take turns: every subscriber is sent every +//! notification, so a test that ran at the same time as another would read the other's +//! notifications too. Window ids are 901 and up. + +use komorebi::SUBSCRIPTION_SOCKETS; +use komorebi::backend::fake::FakeBackend; +use komorebi::backend::set_backend; +use komorebi::process_command::listen_for_commands; +use komorebi::window::Window; +use komorebi::window_manager::WindowManager; +use komorebi::window_manager_event::KWinNotification; +use komorebi::window_manager_event::SystemNotification; +use komorebi::window_manager_event::WindowManagerEvent; +use komorebi::window_manager_event_listener; +use komorebi_client::DATA_DIR; +use komorebi_client::Notification; +use komorebi_client::NotificationEvent; +use komorebi_client::SocketMessage; +use komorebi_client::SubscribeOptions; +use komorebi_client::send_query; +use komorebi_client::subscribe; +use komorebi_client::subscribe_with_options; +use parking_lot::Mutex; +use std::io::ErrorKind; +use std::io::Read; +use std::os::unix::net::UnixListener; +use std::sync::Arc; +use std::sync::OnceLock; +use std::sync::mpsc::Receiver; +use std::time::Duration; + +struct Daemon { + fake: Arc, + wm: Arc>, +} + +impl Daemon { + /// Opens the window `id` and has the window manager handle the event of it being shown + fn show(&self, id: u32) { + self.fake.open_window(id, "foot", "foot", "a terminal"); + self.handle(WindowManagerEvent::Show( + SystemNotification::KWin(KWinNotification::WindowAdded), + Window::from(id), + )); + } + + /// Has the window manager handle the event of the window `id` being minimized + fn minimize(&self, id: u32) { + self.handle(WindowManagerEvent::Minimize( + SystemNotification::KWin(KWinNotification::WindowMinimized), + Window::from(id), + )); + } + + fn handle(&self, event: WindowManagerEvent) { + self.wm.lock().process_event(event).unwrap(); + } +} + +/// Starts the window manager once, and waits for the other tests to be done with it +fn daemon() -> (parking_lot::MutexGuard<'static, ()>, &'static Daemon) { + static TURN: Mutex<()> = parking_lot::const_mutex(()); + static DAEMON: OnceLock = OnceLock::new(); + + let turn = TURN.lock(); + let daemon = DAEMON.get_or_init(|| { + // /tmp: short enough for a unix socket path on macOS + let runtime = format!("/tmp/komorebi-subscribe-test-{}", std::process::id()); + // SAFETY: the first test to get here sets it, before any thread reads the environment + unsafe { std::env::set_var("XDG_RUNTIME_DIR", &runtime) }; + std::fs::create_dir_all(&*DATA_DIR).unwrap(); + + let fake = Arc::new(FakeBackend::with_one_monitor()); + set_backend(fake.clone()); + + let mut wm = WindowManager::new(window_manager_event_listener::event_rx(), None).unwrap(); + wm.init().unwrap(); + let wm = Arc::new(Mutex::new(wm)); + listen_for_commands(wm.clone()); + + Daemon { fake, wm } + }); + + (turn, daemon) +} + +/// A subscriber whose notifications are read on another thread. It unsubscribes when it is +/// dropped, and waits for the window manager to have handled that. +struct Subscriber { + name: &'static str, + notifications: Receiver, +} + +impl Subscriber { + fn new(name: &'static str, listener: UnixListener) -> Self { + let (sender, notifications) = std::sync::mpsc::channel(); + + std::thread::spawn(move || { + for connection in listener.incoming() { + let mut json = String::new(); + connection.unwrap().read_to_string(&mut json).unwrap(); + let notification = serde_json::from_str(&json).unwrap(); + if sender.send(notification).is_err() { + break; + } + } + }); + + Self { + name, + notifications, + } + } + + /// Returns the notifications received until, and with, the first one whose event `last` is + /// true of, waiting up to five seconds for each + fn until(&self, last: impl Fn(&NotificationEvent) -> bool) -> Vec { + let mut received: Vec = vec![]; + loop { + let notification = self + .notifications + .recv_timeout(Duration::from_secs(5)) + .unwrap_or_else(|_| { + panic!("{} received {received:?}, and then nothing", self.name) + }); + let done = last(¬ification.event); + received.push(notification); + if done { + return received; + } + } + } + + /// Returns the next notification, which is the one of the subscription itself + fn subscribed(&self) -> Notification { + let notification = self + .notifications + .recv_timeout(Duration::from_secs(5)) + .expect("the notification of the subscription"); + assert!( + subscribed(self.name)(¬ification.event), + "{notification:?}" + ); + notification + } +} + +impl Drop for Subscriber { + fn drop(&mut self) { + // send_query returns once the window manager has handled the message and closed the + // connection + let unsubscribe = SocketMessage::RemoveSubscriberSocket(self.name.to_string()); + let _ = send_query(&unsubscribe); + } +} + +fn subscribed(name: &str) -> impl Fn(&NotificationEvent) -> bool + '_ { + move |event| match event { + NotificationEvent::Socket(SocketMessage::AddSubscriberSocket(socket)) + | NotificationEvent::Socket(SocketMessage::AddSubscriberSocketWithOptions(socket, _)) => { + socket == name + } + _ => false, + } +} + +fn shown(id: u32) -> impl Fn(&NotificationEvent) -> bool { + move |event| { + matches!( + event, + NotificationEvent::WindowManager(WindowManagerEvent::Show(_, window)) if window.id == id + ) + } +} + +fn minimized(id: u32) -> impl Fn(&NotificationEvent) -> bool { + move |event| { + matches!( + event, + NotificationEvent::WindowManager(WindowManagerEvent::Minimize(_, window)) if window.id == id + ) + } +} + +fn is_state(notification: &Notification) -> bool { + matches!( + notification.event, + NotificationEvent::Socket(SocketMessage::State) + ) +} + +/// Returns true if and only if the first workspace of the first monitor has the window `id` +fn on_the_workspace(notification: &Notification, id: u32) -> bool { + notification.state.monitors.elements()[0].workspaces()[0].contains_window(id) +} + +#[test] +fn a_subscriber_is_sent_the_window_that_was_shown_and_the_state_after_it() +-> Result<(), Box> { + let (_turn, daemon) = daemon(); + let subscriber = Subscriber::new("shown.sock", subscribe("shown.sock")?); + let subscription = subscriber.subscribed(); + assert!(!on_the_workspace(&subscription, 901)); + + daemon.show(901); + + let received = subscriber.until(shown(901)); + assert_eq!(received.len(), 1, "{received:?}"); + assert!(on_the_workspace(&received[0], 901)); + Ok(()) +} + +#[test] +fn a_subscriber_that_filters_state_changes_is_not_sent_what_changed_nothing() +-> Result<(), Box> { + let (_turn, daemon) = daemon(); + let options = SubscribeOptions { + filter_state_changes: true, + }; + let filtered = Subscriber::new( + "filtered.sock", + subscribe_with_options("filtered.sock", options)?, + ); + filtered.subscribed(); + let everything = Subscriber::new("everything.sock", subscribe("everything.sock")?); + everything.subscribed(); + // a subscriber that filters is sent the subscription of another, which changes nothing + assert!(subscribed("everything.sock")( + &filtered.until(|_| true)[0].event + )); + + // asking for the state changes nothing; a window that is shown is always sent, and one that + // is minimized leaves its workspace, which is a change + send_query(&SocketMessage::State)?; + daemon.show(911); + daemon.minimize(911); + + let received = filtered.until(minimized(911)); + assert!(!received.iter().any(is_state), "{received:?}"); + assert!(shown(911)(&received[0].event), "{received:?}"); + assert!(!on_the_workspace(&received[1], 911)); + let received = everything.until(minimized(911)); + assert!(is_state(&received[0]), "{received:?}"); + Ok(()) +} + +#[test] +fn a_subscriber_that_is_gone_is_dropped_and_not_tried_again() +-> Result<(), Box> { + let (_turn, _daemon) = daemon(); + let socket = DATA_DIR.join("gone.sock"); + let gone = subscribe("gone.sock")?; + let mut json = String::new(); + gone.accept()?.0.read_to_string(&mut json)?; + let notification: Notification = serde_json::from_str(&json)?; + assert!(subscribed("gone.sock")(¬ification.event)); + + // the program exits: the socket file is left behind, and nothing listens on it + drop(gone); + assert!(socket.exists()); + send_query(&SocketMessage::State)?; + + assert!(!SUBSCRIPTION_SOCKETS.lock().contains_key("gone.sock")); + assert!(!socket.exists()); + + // a program that listens on the same socket again is sent nothing until it subscribes again + let again = UnixListener::bind(&socket)?; + again.set_nonblocking(true)?; + send_query(&SocketMessage::State)?; + assert_eq!( + again.accept().map(|_| ()).map_err(|error| error.kind()), + Err(ErrorKind::WouldBlock) + ); + std::fs::remove_file(&socket)?; + Ok(()) +} diff --git a/komorebi/src/core/mod.rs b/komorebi/src/core/mod.rs index 91d37c9b..f5c474ae 100644 --- a/komorebi/src/core/mod.rs +++ b/komorebi/src/core/mod.rs @@ -6,8 +6,9 @@ //! import from. [`pathext`] resolves environment variables in configured paths and //! [`config_generation`] holds the rules that match applications by what they are called. //! -//! [`SocketMessage`] is every message the daemon takes on its socket, and [`StateQuery`] every -//! question [`SocketMessage::Query`] can ask. +//! [`SocketMessage`] is every message the daemon takes on its socket, [`StateQuery`] every +//! question [`SocketMessage::Query`] can ask, and [`SubscribeOptions`] what a subscriber can ask +//! of its notifications. use clap::ValueEnum; use color_eyre::eyre; @@ -273,6 +274,40 @@ pub enum StateQuery { Version, } +/// The options of a subscriber, which [`SocketMessage::AddSubscriberSocketWithOptions`] sets +/// +/// A subscriber added with [`SocketMessage::AddSubscriberSocket`] has the default options, and is +/// sent a notification for every event and message the window manager handles. With +/// `filter_state_changes`, it is sent one only when the event or the message changed the +/// [`State`](crate::state::State), and always when a window is shown and when a subscriber is +/// added, so a new subscriber receives the state as soon as it subscribes. +/// +/// # Example +/// +/// ```rust +/// use komorebi::core::SocketMessage; +/// use komorebi::core::SubscribeOptions; +/// +/// let options = SubscribeOptions { +/// filter_state_changes: true, +/// }; +/// let message = SocketMessage::AddSubscriberSocketWithOptions(String::from("bar.sock"), options); +/// assert_eq!( +/// String::from_utf8(message.as_bytes()?)?, +/// r#"{"type":"AddSubscriberSocketWithOptions","content":["bar.sock",{"filter_state_changes":true}]}"#, +/// ); +/// +/// assert!(!SubscribeOptions::default().filter_state_changes); +/// +/// # Ok::<(), Box>(()) +/// ``` +#[derive(Default, Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize)] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +pub struct SubscribeOptions { + /// Only emit notifications when the window manager state has changed + pub filter_state_changes: bool, +} + /// A message to the daemon, as it is written on its socket: JSON, with the variant's name in /// `type` and its arguments, if it has any, in `content`. /// @@ -304,6 +339,11 @@ pub enum SocketMessage { State, MonitorInformation, Query(StateQuery), + AddSubscriberSocket(String), + AddSubscriberSocketWithOptions(String, SubscribeOptions), + RemoveSubscriberSocket(String), + // AddSubscriberPipe(String), + // RemoveSubscriberPipe(String), } impl SocketMessage { diff --git a/komorebi/src/lib.rs b/komorebi/src/lib.rs index d0056428..b3ca5641 100644 --- a/komorebi/src/lib.rs +++ b/komorebi/src/lib.rs @@ -16,10 +16,25 @@ //! The daemon also listens on a socket, `komorebi.sock` in [`struct@DATA_DIR`], for the messages //! in [`core::SocketMessage`]: [`process_command`] reads them and answers, with the window //! manager's [`state::State`] for one. +//! +//! A program can subscribe to the daemon's notifications with a socket of its own in +//! [`struct@DATA_DIR`]: after each event and each message the window manager handles, +//! [`notify_subscribers`] sends every subscriber a [`Notification`] of it, with the state. +use crate::core::SocketMessage; +use crate::core::SubscribeOptions; +use crate::state::State; +use crate::window_manager_event::WindowManagerEvent; use color_eyre::eyre; use lazy_static::lazy_static; +use parking_lot::Mutex; +use serde::Deserialize; +use serde::Serialize; +use std::collections::HashMap; +use std::io::Write; +use std::os::unix::net::UnixStream; use std::path::PathBuf; +use std::sync::Arc; use std::sync::atomic::AtomicI32; #[macro_use] @@ -47,9 +62,9 @@ pub type LibraryError = eyre::Report; lazy_static! { /// Where komorebi keeps the files it needs while it runs, such as the script it loads into - /// KWin and its command socket: `$XDG_RUNTIME_DIR/komorebi`, which is per-user, on tmpfs and - /// removed at logout. Without a runtime directory (a bare ssh session, or the tests on a Mac) - /// it falls back to the local data directory. + /// KWin, its command socket and the sockets of its subscribers: `$XDG_RUNTIME_DIR/komorebi`, + /// which is per-user, on tmpfs and removed at logout. Without a runtime directory (a bare ssh + /// session, or the tests on a Mac) it falls back to the local data directory. pub static ref DATA_DIR: PathBuf = std::env::var_os("XDG_RUNTIME_DIR") .filter(|dir| !dir.is_empty()) .map(|dir| PathBuf::from(dir).join("komorebi")) @@ -64,6 +79,15 @@ lazy_static! { pub static ref LOG_DIR: PathBuf = dirs::data_local_dir() .expect("there is no local data directory") .join("komorebi"); + /// The subscribers: the name of each one's socket, and its path in [`struct@DATA_DIR`] + pub static ref SUBSCRIPTION_SOCKETS: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + /// The options of the subscribers that were added with options, by the name of their socket + /// + /// An entry is not removed with its subscriber: a subscriber that is added again under the + /// same name, without options, keeps the options it had. + pub static ref SUBSCRIPTION_SOCKET_OPTIONS: Arc>> = + Arc::new(Mutex::new(HashMap::new())); } /// The padding, in pixels, between the edge of a monitor's work area and the tiles of its @@ -72,3 +96,169 @@ pub static DEFAULT_WORKSPACE_PADDING: AtomicI32 = AtomicI32::new(5); /// The padding, in pixels, around each tile, for a monitor that sets none of its own. The default /// for this is 5. pub static DEFAULT_CONTAINER_PADDING: AtomicI32 = AtomicI32::new(5); + +/// The event or the message a [`Notification`] is about +/// +/// It is written as the [`WindowManagerEvent`] or the [`SocketMessage`] it carries, each with its +/// `type`, with no tag around it. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +#[serde(untagged)] +pub enum NotificationEvent { + /// An event the compositor reported about a window + WindowManager(WindowManagerEvent), + /// A message the daemon took on its socket + Socket(SocketMessage), +} + +/// A notification that [`notify_subscribers`] sends each subscriber: an event or a message the +/// window manager handled, and its state after it +/// +/// # Example +/// +/// A subscriber reads a notification as JSON: +/// +/// ```rust +/// use komorebi::Notification; +/// use komorebi::NotificationEvent; +/// use komorebi::core::SocketMessage; +/// use komorebi::window_manager_event::WindowManagerEvent; +/// +/// let state = r#"{"monitors":{"elements":[],"focused":0},"work_area_offset":null,"mouse_follows_focus":true}"#; +/// +/// let shown = format!( +/// r#"{{"event":{{"type":"Show","content":[{{"type":"KWin","content":"WindowAdded"}},{{"id":7}}]}},"state":{state}}}"# +/// ); +/// let notification: Notification = serde_json::from_str(&shown)?; +/// assert!(matches!( +/// notification.event, +/// NotificationEvent::WindowManager(WindowManagerEvent::Show(_, window)) if window.id == 7 +/// )); +/// +/// let retiled = format!(r#"{{"event":{{"type":"Retile"}},"state":{state}}}"#); +/// let notification: Notification = serde_json::from_str(&retiled)?; +/// assert!(matches!( +/// notification.event, +/// NotificationEvent::Socket(SocketMessage::Retile) +/// )); +/// assert!(notification.state.mouse_follows_focus); +/// +/// # Ok::<(), Box>(()) +/// ``` +#[derive(Clone, Debug, Serialize, Deserialize)] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +pub struct Notification { + /// The event or the message the window manager handled + pub event: NotificationEvent, + /// The window manager's state after it + pub state: State, +} + +/// Sends `notification` to every subscriber in [`struct@SUBSCRIPTION_SOCKETS`], and removes the +/// subscribers whose socket cannot be connected to +/// +/// Each subscriber is sent the notification as JSON on a new connection to its socket, which is +/// closed after it. A subscriber with `filter_state_changes` in its [`SubscribeOptions`] is sent it +/// only if `state_has_been_modified`, or if the notification is about a window being shown or a +/// subscriber being added. +/// +/// A subscriber that cannot be connected to (its program has exited, or its socket was removed) +/// is removed, with its socket file, and is not tried again: it has to subscribe again. The +/// notification is written without a timeout, with the window manager locked, so a subscriber that +/// listens and never reads can hold up the window manager when a notification does not fit in the +/// socket's buffer. +/// +/// # Errors +/// +/// - The notification cannot be serialized. +/// - The notification cannot be written to a subscriber that was connected to. The subscribers +/// after it are not sent the notification, and none is removed. +/// +/// # Example +/// +/// ```rust +/// # let runtime = format!("/tmp/komorebi-notify-doc-{}", std::process::id()); +/// # std::fs::create_dir_all(format!("{runtime}/komorebi"))?; +/// # unsafe { std::env::set_var("XDG_RUNTIME_DIR", &runtime) }; // the sockets are in it +/// use std::io::Read; +/// use std::os::unix::net::UnixListener; +/// +/// use komorebi::DATA_DIR; +/// use komorebi::Notification; +/// use komorebi::NotificationEvent; +/// use komorebi::SUBSCRIPTION_SOCKETS; +/// use komorebi::core::SocketMessage; +/// use komorebi::notify_subscribers; +/// use komorebi::state::State; +/// +/// // a subscriber that listens, and one whose program has exited +/// let bar = UnixListener::bind(DATA_DIR.join("bar.sock"))?; +/// SUBSCRIPTION_SOCKETS.lock().extend([ +/// (String::from("bar.sock"), DATA_DIR.join("bar.sock")), +/// (String::from("gone.sock"), DATA_DIR.join("gone.sock")), +/// ]); +/// +/// let state: State = serde_json::from_str( +/// r#"{"monitors":{"elements":[],"focused":0},"work_area_offset":null,"mouse_follows_focus":true}"#, +/// )?; +/// let event = NotificationEvent::Socket(SocketMessage::Retile); +/// notify_subscribers(Notification { event, state }, false)?; +/// +/// let mut json = String::new(); +/// bar.accept()?.0.read_to_string(&mut json)?; +/// assert!(json.starts_with(r#"{"event":{"type":"Retile"},"state":{"#), "{json}"); +/// assert!(!SUBSCRIPTION_SOCKETS.lock().contains_key("gone.sock")); +/// +/// # std::fs::remove_dir_all(&runtime)?; +/// # Ok::<(), Box>(()) +/// ``` +pub fn notify_subscribers( + notification: Notification, + state_has_been_modified: bool, +) -> eyre::Result<()> { + let is_override_event = matches!( + notification.event, + NotificationEvent::Socket(SocketMessage::AddSubscriberSocket(_)) + | NotificationEvent::Socket(SocketMessage::AddSubscriberSocketWithOptions(_, _)) + | NotificationEvent::WindowManager(WindowManagerEvent::Show(_, _)) + ); + + let notification = &serde_json::to_string(¬ification)?; + let mut stale_sockets = vec![]; + let mut sockets = SUBSCRIPTION_SOCKETS.lock(); + let options = SUBSCRIPTION_SOCKET_OPTIONS.lock(); + + for (socket, path) in &mut *sockets { + let apply_state_filter = (*options) + .get(socket) + .copied() + .unwrap_or_default() + .filter_state_changes; + + if !apply_state_filter || state_has_been_modified || is_override_event { + match UnixStream::connect(path) { + Ok(mut stream) => { + tracing::debug!("pushed notification to subscriber: {socket}"); + stream.write_all(notification.as_bytes())?; + } + Err(_) => { + stale_sockets.push(socket.clone()); + } + } + } + } + + for socket in stale_sockets { + tracing::warn!("removing stale subscription: {socket}"); + sockets.remove(&socket); + let socket_path = DATA_DIR.join(socket); + if let Err(error) = std::fs::remove_file(&socket_path) { + tracing::error!( + "could not remove stale subscriber socket file at {}: {error}", + socket_path.display() + ) + } + } + + Ok(()) +} diff --git a/komorebi/src/main.rs b/komorebi/src/main.rs index 920791a7..71948819 100644 --- a/komorebi/src/main.rs +++ b/komorebi/src/main.rs @@ -13,6 +13,9 @@ //! //! It logs to standard output and to a file a day in [`struct@LOG_DIR`], which `komorebic log` //! follows. +//! +//! When it stops, it connects to the socket of each subscriber and closes the connection without +//! writing anything, so a subscriber sees the daemon stop as a connection with nothing on it. use clap::Parser; use clap::ValueEnum; @@ -27,6 +30,8 @@ use komorebi::window_manager::WindowManager; use komorebi::window_manager_event_listener; use parking_lot::Mutex; use serde::Deserialize; +use std::net::Shutdown; +use std::os::unix::net::UnixStream; use std::sync::Arc; use tracing_appender::non_blocking::WorkerGuard; use tracing_subscriber::EnvFilter; @@ -191,6 +196,13 @@ fn main() -> eyre::Result<()> { backend::backend().shutdown(); + let sockets = komorebi::SUBSCRIPTION_SOCKETS.lock(); + for path in (*sockets).values() { + if let Ok(stream) = UnixStream::connect(path) { + stream.shutdown(Shutdown::Both)?; + } + } + let socket = DATA_DIR.join("komorebi.sock"); let _ = std::fs::remove_file(socket); diff --git a/komorebi/src/process_command.rs b/komorebi/src/process_command.rs index 37759873..ecd3721c 100644 --- a/komorebi/src/process_command.rs +++ b/komorebi/src/process_command.rs @@ -10,13 +10,24 @@ //! - **MonitorInformation**: the id, the name and the bounds of each monitor, as JSON. //! - **Query**: one value as text, the answer to a [`StateQuery`]. //! -//! The others answer nothing: **Retile** tiles the focused workspace of every monitor again, and -//! **Stop** stops the daemon ([`WindowManager::stop`]). +//! The others answer nothing: **Retile** tiles the focused workspace of every monitor again, +//! **AddSubscriberSocket** and **AddSubscriberSocketWithOptions** add a subscriber by the name of +//! its socket in the data directory, **RemoveSubscriberSocket** removes one, and **Stop** stops the +//! daemon ([`WindowManager::stop`]). +//! +//! Once a message has been handled, every subscriber is sent a notification of it +//! ([`notify_subscribers`]). +use crate::DATA_DIR; +use crate::Notification; +use crate::NotificationEvent; +use crate::SUBSCRIPTION_SOCKET_OPTIONS; +use crate::SUBSCRIPTION_SOCKETS; use crate::core::Layout; use crate::core::SocketMessage; use crate::core::StateQuery; use crate::monitor::MonitorInformation; +use crate::notify_subscribers; use crate::state::State; use crate::window_manager::WindowManager; use color_eyre::eyre; @@ -77,7 +88,10 @@ pub fn listen_for_commands(wm: Arc>) { } impl WindowManager { - /// Handles one message, and writes the answer to a question to `reply` + /// Handles one message, writes the answer to a question to `reply`, and sends the subscribers + /// a notification of the message + /// + /// A message that fails is not notified. /// /// # Errors /// @@ -85,6 +99,8 @@ impl WindowManager { /// - The window manager fails to retile, for one of the reasons /// [`WindowManager::retile_all`] gives. /// - The answer cannot be written to `reply`. + /// - `Stop` fails, for the reason [`WindowManager::stop`] gives. + /// - The notification cannot be sent, for one of the reasons [`notify_subscribers`] gives. #[tracing::instrument(skip(self, reply, message))] pub fn process_command( &mut self, @@ -93,6 +109,10 @@ impl WindowManager { ) -> eyre::Result<()> { tracing::info!("processing command: {message}"); + #[allow(clippy::useless_asref)] + // We don't have From implemented for &mut WindowManager + let initial_state = State::from(self.as_ref()); + match message { SocketMessage::Retile => self.retile_all()?, SocketMessage::State => { @@ -161,6 +181,23 @@ impl WindowManager { reply.write_all(response.as_bytes())?; } + SocketMessage::AddSubscriberSocket(ref socket) => { + let mut sockets = SUBSCRIPTION_SOCKETS.lock(); + let socket_path = DATA_DIR.join(socket); + sockets.insert(socket.clone(), socket_path); + } + SocketMessage::AddSubscriberSocketWithOptions(ref socket, options) => { + let mut sockets = SUBSCRIPTION_SOCKETS.lock(); + let socket_path = DATA_DIR.join(socket); + sockets.insert(socket.clone(), socket_path); + + let mut socket_options = SUBSCRIPTION_SOCKET_OPTIONS.lock(); + socket_options.insert(socket.clone(), options); + } + SocketMessage::RemoveSubscriberSocket(ref socket) => { + let mut sockets = SUBSCRIPTION_SOCKETS.lock(); + sockets.remove(socket); + } SocketMessage::Stop => { self.stop()?; } @@ -168,6 +205,14 @@ impl WindowManager { self.update_known_window_ids(); + notify_subscribers( + Notification { + event: NotificationEvent::Socket(message.clone()), + state: self.as_ref().into(), + }, + initial_state.has_been_modified(self.as_ref()), + )?; + tracing::info!("processed command: {message}"); Ok(()) diff --git a/komorebi/src/process_event.rs b/komorebi/src/process_event.rs index 8f051fd3..c497b131 100644 --- a/komorebi/src/process_event.rs +++ b/komorebi/src/process_event.rs @@ -15,9 +15,16 @@ //! The daemon hands events to the window manager on a thread of their own, //! [`listen_for_events`], one at a time and in the order the backend raised them. //! +//! Once an event about a managed window has been handled, every subscriber is sent a notification +//! of it ([`notify_subscribers`]). +//! //! [`Window::should_manage`]: crate::window::Window::should_manage +use crate::Notification; +use crate::NotificationEvent; use crate::backend::backend; +use crate::notify_subscribers; +use crate::state::State; use crate::window_manager::WindowManager; use crate::window_manager_event::WindowManagerEvent; use color_eyre::eyre; @@ -63,6 +70,10 @@ fn should_refocus_after_destroy(foreground: Option) -> bool { impl WindowManager { /// Handle an event from the compositor, and log how long it took + /// + /// The subscribers are sent a notification of an event that was handled + /// ([`notify_subscribers`]), and a notification that cannot be sent is an error. An event + /// that fails, or is about a window komorebi does not manage, is not notified. pub fn process_event(&mut self, event: WindowManagerEvent) -> eyre::Result<()> { let started = std::time::Instant::now(); let result = self.handle_event(event); @@ -97,6 +108,10 @@ impl WindowManager { // last one finished self.update_known_window_ids(); + #[allow(clippy::useless_asref)] + // We don't have From implemented for &mut WindowManager + let initial_state = State::from(self.as_ref()); + match event { WindowManagerEvent::FocusChange(_, _) => { let mut is_on_current_workspace = false; @@ -189,6 +204,14 @@ impl WindowManager { self.update_known_window_ids(); + notify_subscribers( + Notification { + event: NotificationEvent::WindowManager(event), + state: self.as_ref().into(), + }, + initial_state.has_been_modified(self.as_ref()), + )?; + Ok(()) } } diff --git a/komorebi/src/state.rs b/komorebi/src/state.rs index 885dc280..37cbcc23 100644 --- a/komorebi/src/state.rs +++ b/komorebi/src/state.rs @@ -1,9 +1,12 @@ //! [`State`], the window manager as the daemon shows it: every monitor, with its workspaces, their //! containers and the windows in them, and the settings that apply to all of them. //! -//! The daemon writes it as JSON in answer to [`SocketMessage::State`]. +//! The daemon writes it as JSON in answer to [`SocketMessage::State`], and sends it to its +//! subscribers in every [`Notification`]. [`State::has_been_modified`] decides whether an event or +//! a message changed it, for the subscribers that want a notification only then. //! //! [`SocketMessage::State`]: crate::core::SocketMessage::State +//! [`Notification`]: crate::Notification use crate::core::Rect; use crate::monitor::Monitor; @@ -98,3 +101,29 @@ impl From<&WindowManager> for State { } } } + +impl State { + /// Returns true if and only if the window manager's monitors (with their workspaces, + /// containers and windows), its work area offset or its mouse follows focus setting differ + /// from this state's + /// + /// A window is compared by its id only, so a window that moved or was renamed is not a + /// change. + pub fn has_been_modified(&self, wm: &WindowManager) -> bool { + let new = Self::from(wm); + + if self.monitors != new.monitors { + return true; + } + + if self.work_area_offset != new.work_area_offset { + return true; + } + + if self.mouse_follows_focus != new.mouse_follows_focus { + return true; + } + + false + } +} diff --git a/komorebi/src/window_manager.rs b/komorebi/src/window_manager.rs index b662efce..4e99ad3d 100644 --- a/komorebi/src/window_manager.rs +++ b/komorebi/src/window_manager.rs @@ -14,6 +14,7 @@ use crate::DATA_DIR; use crate::LibraryError; +use crate::SUBSCRIPTION_SOCKETS; use crate::backend::backend; use crate::container::Container; use crate::core::Rect; @@ -28,7 +29,9 @@ use color_eyre::eyre::OptionExt; use crossbeam_channel::Receiver; use std::collections::HashMap; use std::io::ErrorKind; +use std::net::Shutdown; use std::os::unix::net::UnixListener; +use std::os::unix::net::UnixStream; use std::path::PathBuf; #[derive(Debug)] @@ -49,6 +52,12 @@ pub struct WindowManager { impl_ring_elements!(WindowManager, Monitor); +impl AsRef for WindowManager { + fn as_ref(&self) -> &Self { + self + } +} + impl WindowManager { /// Returns a window manager with no monitors, which handles the events that arrive on /// `incoming` and listens on `custom_socket_path`, or on `komorebi.sock` in @@ -196,12 +205,24 @@ impl WindowManager { self.update_focused_workspace(false, false) } - /// Stops the daemon: removes the command socket, lets the backend give up what it set up in - /// the compositor, and exits the process with status 0 + /// Stops the daemon: connects to every subscriber and closes the connection without writing + /// anything, removes the command socket, lets the backend give up what it set up in the + /// compositor, and exits the process with status 0 + /// + /// # Errors + /// + /// A connection to a subscriber cannot be shut down. The daemon does not stop. #[tracing::instrument(skip(self))] pub fn stop(&mut self) -> eyre::Result<()> { tracing::info!("received stop command, terminating process"); + let sockets = SUBSCRIPTION_SOCKETS.lock(); + for path in (*sockets).values() { + if let Ok(stream) = UnixStream::connect(path) { + stream.shutdown(Shutdown::Both)?; + } + } + let socket = DATA_DIR.join("komorebi.sock"); let _ = std::fs::remove_file(socket); diff --git a/komorebic/src/main.rs b/komorebic/src/main.rs index 5ef74b49..2e0c4805 100644 --- a/komorebic/src/main.rs +++ b/komorebic/src/main.rs @@ -5,6 +5,10 @@ //! is written and the answer is printed. The messages are [`komorebi_client::SocketMessage`]s, //! sent with [`komorebi_client::send_message`] and [`komorebi_client::send_query`]. //! +//! `subscribe-socket` and `unsubscribe-socket` add and remove a subscriber: a program that listens +//! on a socket in the daemon's data directory, to which the daemon sends a notification after each +//! event and each message it handles. +//! //! The others work without the daemon: `start` runs `komorebi` in the background, `log` follows //! its log, `data-directory` shows where its socket is, and `completions` prints a completion //! script for a shell. The hidden `docgen` writes the help of every subcommand as Markdown, wrapped @@ -47,6 +51,18 @@ gen_enum_subcommand_args! { Query: StateQuery, } +#[derive(Parser)] +struct SubscribeSocket { + /// Name of the socket to send event notifications to + socket: String, +} + +#[derive(Parser)] +struct UnsubscribeSocket { + /// Name of the socket to stop sending event notifications to + socket: String, +} + #[derive(Parser)] #[allow(clippy::struct_excessive_bools)] struct Start { @@ -100,6 +116,20 @@ enum SubCommand { /// Query the current window manager state #[clap(arg_required_else_help = true)] Query(Query), + /// Subscribe to komorebi events using a Unix Domain Socket + #[clap(arg_required_else_help = true)] + SubscribeSocket(SubscribeSocket), + /// Unsubscribe from komorebi events + #[clap(arg_required_else_help = true)] + UnsubscribeSocket(UnsubscribeSocket), + // /// Subscribe to komorebi events using a Named Pipe + // #[clap(arg_required_else_help = true)] + // #[clap(alias = "subscribe")] + // SubscribePipe(SubscribePipe), + // /// Unsubscribe from komorebi events + // #[clap(arg_required_else_help = true)] + // #[clap(alias = "unsubscribe")] + // UnsubscribePipe(UnsubscribePipe), /// Tail komorebi's process logs (cancel with Ctrl-C) Log, /// Force the retiling of all managed windows @@ -233,6 +263,12 @@ fn main() -> eyre::Result<()> { SubCommand::MonitorInformation => { print_query(&SocketMessage::MonitorInformation); } + SubCommand::SubscribeSocket(args) => { + send_message(&SocketMessage::AddSubscriberSocket(args.socket))?; + } + SubCommand::UnsubscribeSocket(args) => { + send_message(&SocketMessage::RemoveSubscriberSocket(args.socket))?; + } } Ok(()) diff --git a/komorebic/tests/cli.rs b/komorebic/tests/cli.rs index ce285a94..c50a99a5 100644 --- a/komorebic/tests/cli.rs +++ b/komorebic/tests/cli.rs @@ -121,6 +121,29 @@ fn stop_and_retile_send_their_message_and_print_nothing() { } } +#[test] +fn subscribe_socket_and_unsubscribe_socket_send_the_name_they_are_given() { + for (subcommand, message) in [ + ( + "subscribe-socket", + r#"{"type":"AddSubscriberSocket","content":"bar.sock"}"#, + ), + ( + "unsubscribe-socket", + r#"{"type":"RemoveSubscriberSocket","content":"bar.sock"}"#, + ), + ] { + let runtime = Runtime::new(); + let daemon = runtime.serve(""); + + let output = runtime.komorebic(&[subcommand, "bar.sock"]); + + assert!(output.status.success(), "{subcommand}: {output:?}"); + assert_eq!(daemon.join().unwrap(), message, "{subcommand}"); + assert_eq!(stdout(&output), "", "{subcommand}"); + } +} + #[test] fn without_a_daemon_a_question_fails() { let runtime = Runtime::new(); @@ -169,6 +192,8 @@ fn docgen_writes_the_help_of_every_subcommand_in_a_text_fence() { "start.md", "state.md", "stop.md", + "subscribe-socket.md", + "unsubscribe-socket.md", ] );