Merge pull request #164 from pinnacle-comp/signal_minimal

Scaffold a signal system

Cargo.lock

@@ -1622,6 +1622,7 @@ dependencies = [
  "pinnacle-api-defs",
  "pinnacle-api-macros",
  "tokio",
+ "tokio-stream",
  "tonic",
  "tower",
  "xkbcommon",

Cargo.toml

@@ -8,6 +8,7 @@ repository = "https://github.com/pinnacle-comp/pinnacle/"
 
 [workspace.dependencies]
 tokio = { version = "1.36.0", features = ["macros", "rt-multi-thread"]}
+tokio-stream = { version = "0.1.14", features = ["net"] }
 
 prost = "0.12.3"
 tonic = "0.11.0"
@@ -61,7 +62,7 @@ tonic = { workspace = true }
 tonic-reflection = { workspace = true }
 
 tokio = { workspace = true, features = ["process", "io-util", "signal"] }
-tokio-stream = { version = "0.1.14", features = ["net"] }
+tokio-stream = { workspace = true }
 
 bitflags = "2.4.2"
 pinnacle-api-defs = { workspace = true }

api/lua/pinnacle-api-dev-1.rockspec

@@ -24,7 +24,9 @@ build = {
         ["pinnacle.output"] = "pinnacle/output.lua",
         ["pinnacle.process"] = "pinnacle/process.lua",
         ["pinnacle.tag"] = "pinnacle/tag.lua",
+        ["pinnacle.tag.layout"] = "pinnacle/tag/layout.lua",
         ["pinnacle.window"] = "pinnacle/window.lua",
         ["pinnacle.util"] = "pinnacle/util.lua",
+        ["pinnacle.signal"] = "pinnacle/signal.lua",
     },
 }

api/lua/pinnacle/grpc/client.lua

@@ -5,6 +5,7 @@
 local socket = require("cqueues.socket")
 local headers = require("http.headers")
 local h2_connection = require("http.h2_connection")
+local protobuf = require("pinnacle.grpc.protobuf")
 local pb = require("pb")
 
 ---@nodoc
@@ -40,6 +41,10 @@ end
 ---@class H2Connection
 ---@field new_stream function
 
+---@class H2Stream
+---@field write_chunk function
+---@field shutdown function
+
 ---@nodoc
 ---@class Client
 ---@field conn H2Connection
@@ -76,12 +81,7 @@ function client.unary_request(grpc_request_params)
     local response_type = grpc_request_params.response_type or "google.protobuf.Empty"
     local data = grpc_request_params.data
 
-    local encoded_protobuf = assert(pb.encode(request_type, data), "wrong table schema")
-
-    local packed_prefix = string.pack("I1", 0)
-    local payload_len = string.pack(">I4", encoded_protobuf:len())
-
-    local body = packed_prefix .. payload_len .. encoded_protobuf
+    local body = protobuf.encode(request_type, data)
 
     stream:write_headers(create_request_headers(service, method), false)
     stream:write_chunk(body, true)
@@ -126,18 +126,7 @@ function client.server_streaming_request(grpc_request_params, callback)
     local response_type = grpc_request_params.response_type or "google.protobuf.Empty"
     local data = grpc_request_params.data
 
-    local success, obj = pcall(pb.encode, request_type, data)
-    if not success then
-        print("failed to encode:", obj, "for", service, method, request_type, response_type)
-        os.exit(1)
-    end
-
-    local encoded_protobuf = obj
-
-    local packed_prefix = string.pack("I1", 0)
-    local payload_len = string.pack(">I4", encoded_protobuf:len())
-
-    local body = packed_prefix .. payload_len .. encoded_protobuf
+    local body = protobuf.encode(request_type, data)
 
     stream:write_headers(create_request_headers(service, method), false)
     stream:write_chunk(body, true)
@@ -171,4 +160,54 @@ function client.server_streaming_request(grpc_request_params, callback)
     end)
 end
 
+---@nodoc
+---@param grpc_request_params GrpcRequestParams
+---@param callback fun(response: table)
+---
+---@return H2Stream
+function client.bidirectional_streaming_request(grpc_request_params, callback)
+    local stream = client.conn:new_stream()
+
+    local service = grpc_request_params.service
+    local method = grpc_request_params.method
+    local request_type = grpc_request_params.request_type
+    local response_type = grpc_request_params.response_type or "google.protobuf.Empty"
+    local data = grpc_request_params.data
+
+    local body = protobuf.encode(request_type, data)
+
+    stream:write_headers(create_request_headers(service, method), false)
+    stream:write_chunk(body, false)
+
+    -- TODO: check response headers for errors
+    local _ = stream:get_headers()
+
+    client.loop:wrap(function()
+        for response_body in stream:each_chunk() do
+            -- Skip the 1-byte compressed flag and the 4-byte message length
+            ---@diagnostic disable-next-line: redefined-local
+            local response_body = response_body:sub(6)
+
+            ---@diagnostic disable-next-line: redefined-local
+            local success, obj = pcall(pb.decode, response_type, response_body)
+            if not success then
+                print(obj)
+                os.exit(1)
+            end
+
+            local response = obj
+            callback(response)
+        end
+
+        local trailers = stream:get_headers()
+        if trailers then
+            for name, value, never_index in trailers:each() do
+                print(name, value, never_index)
+            end
+        end
+    end)
+
+    return stream
+end
+
 return client

api/lua/pinnacle/grpc/protobuf.lua

@@ -17,6 +17,7 @@ function protobuf.build_protos()
         PINNACLE_PROTO_DIR .. "/pinnacle/output/" .. version .. "/output.proto",
         PINNACLE_PROTO_DIR .. "/pinnacle/process/" .. version .. "/process.proto",
         PINNACLE_PROTO_DIR .. "/pinnacle/window/" .. version .. "/window.proto",
+        PINNACLE_PROTO_DIR .. "/pinnacle/signal/" .. version .. "/signal.proto",
     }
 
     local cmd = "protoc --descriptor_set_out=/tmp/pinnacle.pb --proto_path=" .. PINNACLE_PROTO_DIR .. " "
@@ -38,4 +39,26 @@ function protobuf.build_protos()
     pb.option("enum_as_value")
 end
 
+---@nodoc
+---Encode the given `data` as the protobuf `type`.
+---@param type string The absolute protobuf type
+---@param data table The table of data, conforming to its protobuf definition
+---@return string buffer The encoded buffer
+function protobuf.encode(type, data)
+    local success, obj = pcall(pb.encode, type, data)
+    if not success then
+        print("failed to encode:", obj, "type:", type)
+        os.exit(1)
+    end
+
+    local encoded_protobuf = obj
+
+    local packed_prefix = string.pack("I1", 0)
+    local payload_len = string.pack(">I4", encoded_protobuf:len())
+
+    local body = packed_prefix .. payload_len .. encoded_protobuf
+
+    return body
+end
+
 return protobuf

api/lua/pinnacle/output.lua

@@ -159,13 +159,56 @@ function output.connect_for_all(callback)
         callback(handle)
     end
 
-    client.server_streaming_request(build_grpc_request_params("ConnectForAll", {}), function(response)
-        local output_name = response.output_name
-        local handle = output_handle.new(output_name)
-        callback(handle)
-    end)
+    output.connect_signal({
+        connect = callback,
+    })
 end
 
+local signal_name_to_SignalName = {
+    connect = "OutputConnect",
+}
+
+---@class OutputSignal Signals related to output events.
+---@field connect fun(output: OutputHandle)? An output was connected. FIXME: This currently does not fire for outputs that have been previously connected and disconnected.
+
+---Connect to an output signal.
+---
+---The compositor sends signals about various events. Use this function to run a callback when
+---some output signal occurs.
+---
+---This function returns a table of signal handles with each handle stored at the same key used
+---to connect to the signal. See `SignalHandles` for more information.
+---
+---# Example
+---```lua
+---Output.connect_signal({
+---    connect = function(output)
+---        print("New output connected:", output.name)
+---    end
+---})
+---```
+---
+---@param signals OutputSignal The signal you want to connect to
+---
+---@return SignalHandles signal_handles Handles to every signal you connected to wrapped in a table, with keys being the same as the connected signal.
+---
+---@see SignalHandles.disconnect_all - To disconnect from these signals
+function output.connect_signal(signals)
+    ---@diagnostic disable-next-line: invisible
+    local handles = require("pinnacle.signal").handles.new({})
+
+    for signal, callback in pairs(signals) do
+        require("pinnacle.signal").add_callback(signal_name_to_SignalName[signal], callback)
+        ---@diagnostic disable-next-line: invisible
+        local handle = require("pinnacle.signal").handle.new(signal_name_to_SignalName[signal], callback)
+        handles[signal] = handle
+    end
+
+    return handles
+end
+
+---------------------------------------------------------------------
+
 ---Set the location of this output in the global space.
 ---
 ---On startup, Pinnacle will lay out all connected outputs starting at (0, 0)

api/lua/pinnacle/signal.lua

@@ -0,0 +1,281 @@
+-- This Source Code Form is subject to the terms of the Mozilla Public
+-- License, v. 2.0. If a copy of the MPL was not distributed with this
+-- file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+local client = require("pinnacle.grpc.client")
+
+---The protobuf absolute path prefix
+local prefix = "pinnacle.signal." .. client.version .. "."
+local service = prefix .. "SignalService"
+
+---@type table<string, { request_type: string?, response_type: string? }>
+---@enum (key) SignalServiceMethod
+local rpc_types = {
+    OutputConnect = {
+        response_type = "OutputConnectResponse",
+    },
+    Layout = {
+        response_type = "LayoutResponse",
+    },
+    WindowPointerEnter = {
+        response_type = "WindowPointerEnterResponse",
+    },
+    WindowPointerLeave = {
+        response_type = "WindowPointerLeaveResponse",
+    },
+}
+
+---Build GrpcRequestParams
+---@param method SignalServiceMethod
+---@param data table
+---@return GrpcRequestParams
+local function build_grpc_request_params(method, data)
+    local req_type = rpc_types[method].request_type
+    local resp_type = rpc_types[method].response_type
+
+    ---@type GrpcRequestParams
+    return {
+        service = service,
+        method = method,
+        request_type = req_type and prefix .. req_type or prefix .. method .. "Request",
+        response_type = resp_type and prefix .. resp_type,
+        data = data,
+    }
+end
+
+local stream_control = {
+    UNSPECIFIED = 0,
+    READY = 1,
+    DISCONNECT = 2,
+}
+
+-- TODO: rewrite ldoc_gen so you don't have to stick @nodoc everywhere
+
+---@nodoc
+---@type table<SignalServiceMethod, { sender: H2Stream?, callbacks: function[], on_response: fun(response: table) }>
+local signals = {
+    OutputConnect = {
+        ---@nodoc
+        ---@type H2Stream?
+        sender = nil,
+        ---@nodoc
+        ---@type (fun(output: OutputHandle))[]
+        callbacks = {},
+        ---@nodoc
+        ---@type fun(response: table)
+        on_response = nil,
+    },
+    Layout = {
+        ---@nodoc
+        ---@type H2Stream?
+        sender = nil,
+        ---@nodoc
+        ---@type (fun(tag: TagHandle, windows: WindowHandle[]))[]
+        callbacks = {},
+        ---@nodoc
+        ---@type fun(response: table)
+        on_response = nil,
+    },
+    WindowPointerEnter = {
+        ---@nodoc
+        ---@type H2Stream?
+        sender = nil,
+        ---@nodoc
+        ---@type (fun(window: WindowHandle))[]
+        callbacks = {},
+        ---@nodoc
+        ---@type fun(response: table)
+        on_response = nil,
+    },
+    WindowPointerLeave = {
+        ---@nodoc
+        ---@type H2Stream?
+        sender = nil,
+        ---@nodoc
+        ---@type (fun(window: WindowHandle))[]
+        callbacks = {},
+        ---@nodoc
+        ---@type fun(response: table)
+        on_response = nil,
+    },
+}
+
+signals.OutputConnect.on_response = function(response)
+    ---@diagnostic disable-next-line: invisible
+    local handle = require("pinnacle.output").handle.new(response.output_name)
+    for _, callback in ipairs(signals.OutputConnect.callbacks) do
+        callback(handle)
+    end
+end
+
+signals.Layout.on_response = function(response)
+    ---@diagnostic disable-next-line: invisible
+    local window_handles = require("pinnacle.window").handle.new_from_table(response.window_ids or {})
+    ---@diagnostic disable-next-line: invisible
+    local tag_handle = require("pinnacle.tag").handle.new(response.tag_id)
+
+    for _, callback in ipairs(signals.Layout.callbacks) do
+        callback(tag_handle, window_handles)
+    end
+end
+
+signals.WindowPointerEnter.on_response = function(response)
+    ---@diagnostic disable-next-line: invisible
+    local window_handle = require("pinnacle.window").handle.new(response.window_id)
+
+    for _, callback in ipairs(signals.WindowPointerEnter.callbacks) do
+        callback(window_handle)
+    end
+end
+
+signals.WindowPointerLeave.on_response = function(response)
+    ---@diagnostic disable-next-line: invisible
+    local window_handle = require("pinnacle.window").handle.new(response.window_id)
+
+    for _, callback in ipairs(signals.WindowPointerLeave.callbacks) do
+        callback(window_handle)
+    end
+end
+
+-----------------------------------------------------------------------------
+
+---@nodoc
+---@class SignalHandleModule
+local signal_handle = {}
+
+---@classmod
+---A handle to a connected signal that can be used to disconnect the provided callback.
+---
+---@class SignalHandle
+---@field private signal SignalServiceMethod
+---@field private callback function The callback you connected
+local SignalHandle = {}
+
+---@nodoc
+---@class SignalHandlesModule
+local signal_handles = {}
+
+---A collection of `SignalHandle`s retreived through a `connect_signal` function.
+---@classmod
+---@class SignalHandles
+local SignalHandles = {}
+
+---@nodoc
+---@class Signal
+---@field private handle SignalHandleModule
+---@field private handles SignalHandlesModule
+local signal = {}
+signal.handle = signal_handle
+signal.handles = signal_handles
+
+---Disconnect the provided callback from this signal.
+function SignalHandle:disconnect()
+    local cb_index = nil
+    for i, cb in ipairs(signals[self.signal].callbacks) do
+        if cb == self.callback then
+            cb_index = i
+            break
+        end
+    end
+
+    if cb_index then
+        table.remove(signals[self.signal].callbacks, cb_index)
+    end
+
+    if #signals[self.signal].callbacks == 0 then
+        signal.disconnect(self.signal)
+    end
+end
+
+---@nodoc
+---@return SignalHandle
+function signal_handle.new(request, callback)
+    ---@type SignalHandle
+    local self = {
+        signal = request,
+        callback = callback,
+    }
+    setmetatable(self, { __index = SignalHandle })
+    return self
+end
+
+---Disconnect the callbacks from all the signal connections that are stored in this handle collection.
+---
+---@param self table<string, SignalHandle>
+function SignalHandles:disconnect_all()
+    for _, sig in pairs(self) do
+        sig:disconnect()
+    end
+end
+
+---@nodoc
+---@param signal_hdls table<string, SignalHandle>
+---@return SignalHandles
+function signal_handles.new(signal_hdls)
+    ---@type SignalHandles
+    local self = signal_hdls
+    setmetatable(self, { __index = SignalHandles })
+    return self
+end
+
+---@nodoc
+---@param request SignalServiceMethod
+---@param callback function
+function signal.add_callback(request, callback)
+    if #signals[request].callbacks == 0 then
+        signal.connect(request, signals[request].on_response)
+    end
+
+    table.insert(signals[request].callbacks, callback)
+end
+
+---@nodoc
+---@param request SignalServiceMethod
+---@param callback fun(response: table)
+function signal.connect(request, callback)
+    local stream = client.bidirectional_streaming_request(
+        build_grpc_request_params(request, {
+            control = stream_control.READY,
+        }),
+        function(response)
+            callback(response)
+
+            if signals[request].sender then
+                local chunk = require("pinnacle.grpc.protobuf").encode(prefix .. request .. "Request", {
+                    control = stream_control.READY,
+                })
+
+                local success, err = pcall(signals[request].sender.write_chunk, signals[request].sender, chunk)
+
+                if not success then
+                    print("error sending to stream:", err)
+                    os.exit(1)
+                end
+            end
+        end
+    )
+
+    signals[request].sender = stream
+end
+
+---@nodoc
+---This should only be called when call callbacks for the signal are removed
+---@param request SignalServiceMethod
+function signal.disconnect(request)
+    if signals[request].sender then
+        local chunk = require("pinnacle.grpc.protobuf").encode(prefix .. request .. "Request", {
+            control = stream_control.DISCONNECT,
+        })
+
+        local success, err = pcall(signals[request].sender.write_chunk, signals[request].sender, chunk)
+        if not success then
+            print("error sending to stream:", err)
+            os.exit(1)
+        end
+
+        signals[request].sender:shutdown()
+        signals[request].sender = nil
+    end
+end
+
+return signal

api/lua/pinnacle/tag.lua

@@ -234,8 +234,8 @@ end
 ---layout_cycler.next() -- Layout is now "dwindle"
 ---layout_cycler.next() -- Layout is now "corner_top_left"
 ---layout_cycler.next() -- Layout is now "corner_top_right"
+---layout_cycler.next() -- Layout is now "master_stack"
 ---layout_cycler.next() -- Layout is now "dwindle"
----layout_cycler.next() -- Layout is now "corner_top_right"
 ---
 --- -- Cycling on another output
 ---layout_cycler.next(Output.get_by_name("eDP-1"))
@@ -319,6 +319,49 @@ function tag.new_layout_cycler(layouts)
     }
 end
 
+local signal_name_to_SignalName = {
+    layout = "Layout",
+}
+
+---@class TagSignal Signals related to tag events.
+---@field layout fun(tag: TagHandle, windows: WindowHandle[])? The compositor requested a layout of the given tiled windows. You'll also receive the first active tag.
+
+---Connect to a tag signal.
+---
+---The compositor sends signals about various events. Use this function to run a callback when
+---some tag signal occurs.
+---
+---This function returns a table of signal handles with each handle stored at the same key used
+---to connect to the signal. See `SignalHandles` for more information.
+---
+---# Example
+---```lua
+---Tag.connect_signal({
+---    layout = function(tag, windows)
+---        print("Compositor requested a layout")
+---    end
+---})
+---```
+---
+---@param signals TagSignal The signal you want to connect to
+---
+---@return SignalHandles signal_handles Handles to every signal you connected to wrapped in a table, with keys being the same as the connected signal.
+---
+---@see SignalHandles.disconnect_all - To disconnect from these signals
+function tag.connect_signal(signals)
+    ---@diagnostic disable-next-line: invisible
+    local handles = require("pinnacle.signal").handles.new({})
+
+    for signal, callback in pairs(signals) do
+        require("pinnacle.signal").add_callback(signal_name_to_SignalName[signal], callback)
+        ---@diagnostic disable-next-line: invisible
+        local handle = require("pinnacle.signal").handle.new(signal_name_to_SignalName[signal], callback)
+        handles[signal] = handle
+    end
+
+    return handles
+end
+
 ---Remove this tag.
 ---
 ---### Example

api/lua/pinnacle/tag/layout.lua

@@ -0,0 +1,4 @@
+---@class LayoutModule
+local layout = {}
+
+return layout

api/lua/pinnacle/window.lua

@@ -69,7 +69,9 @@ local WindowHandle = {}
 ---This module helps you deal with setting windows to fullscreen and maximized, setting their size,
 ---moving them between tags, and various other actions.
 ---@class Window
+---@field private handle WindowHandleModule
 local window = {}
+window.handle = window_handle
 
 ---Get all windows.
 ---
@@ -309,6 +311,53 @@ function window.add_window_rule(rule)
     }))
 end
 
+local signal_name_to_SignalName = {
+    pointer_enter = "WindowPointerEnter",
+    pointer_leave = "WindowPointerLeave",
+}
+
+---@class WindowSignal Signals related to compositor events.
+---@field pointer_enter fun(window: WindowHandle)? The pointer entered a window.
+---@field pointer_leave fun(window: WindowHandle)? The pointer left a window.
+
+---Connect to a window signal.
+---
+---The compositor sends signals about various events. Use this function to run a callback when
+---some window signal occurs.
+---
+---This function returns a table of signal handles with each handle stored at the same key used
+---to connect to the signal. See `SignalHandles` for more information.
+---
+---# Example
+---```lua
+---Window.connect_signal({
+---    pointer_enter = function(window)
+---        print("Pointer entered", window:class())
+---    end
+---})
+---```
+---
+---@param signals WindowSignal The signal you want to connect to
+---
+---@return SignalHandles signal_handles Handles to every signal you connected to wrapped in a table, with keys being the same as the connected signal.
+---
+---@see SignalHandles.disconnect_all - To disconnect from these signals
+function window.connect_signal(signals)
+    ---@diagnostic disable-next-line: invisible
+    local handles = require("pinnacle.signal").handles.new({})
+
+    for signal, callback in pairs(signals) do
+        require("pinnacle.signal").add_callback(signal_name_to_SignalName[signal], callback)
+        ---@diagnostic disable-next-line: invisible
+        local handle = require("pinnacle.signal").handle.new(signal_name_to_SignalName[signal], callback)
+        handles[signal] = handle
+    end
+
+    return handles
+end
+
+------------------------------------------------------------------------
+
 ---Send a close request to this window.
 ---
 ---### Example

api/protocol/pinnacle/output/v0alpha1/output.proto

@@ -10,11 +10,6 @@ message SetLocationRequest {
   optional int32 y = 3;
 }
 
-message ConnectForAllRequest {}
-message ConnectForAllResponse {
-  optional string output_name = 1;
-}
-
 message GetRequest {}
 message GetResponse {
   repeated string output_names = 1;
@@ -41,7 +36,6 @@ message GetPropertiesResponse {
 
 service OutputService {
   rpc SetLocation(SetLocationRequest) returns (google.protobuf.Empty);
-  rpc ConnectForAll(ConnectForAllRequest) returns (stream ConnectForAllResponse);
   rpc Get(GetRequest) returns (GetResponse);
   rpc GetProperties(GetPropertiesRequest) returns (GetPropertiesResponse);
 }

api/protocol/pinnacle/signal/v0alpha1/signal.proto

@@ -0,0 +1,51 @@
+syntax = "proto2";
+
+package pinnacle.signal.v0alpha1;
+
+enum StreamControl {
+  STREAM_CONTROL_UNSPECIFIED = 0;
+  // The client is ready to receive the next signal.
+  STREAM_CONTROL_READY = 1;
+  // The client wishes to disconnect a signal connection.
+  STREAM_CONTROL_DISCONNECT = 2;
+}
+
+message OutputConnectRequest {
+  optional StreamControl control = 1;
+}
+message OutputConnectResponse {
+  optional string output_name = 1;
+}
+
+message LayoutRequest {
+  optional StreamControl control = 1;
+}
+message LayoutResponse {
+  // The windows that need to be laid out.
+  repeated uint32 window_ids = 1;
+  // The tag that is being laid out.
+  optional uint32 tag_id = 2;
+}
+
+message WindowPointerEnterRequest {
+  optional StreamControl control = 1;
+}
+message WindowPointerEnterResponse {
+  // The window that the pointer entered.
+  optional uint32 window_id = 1;
+}
+
+message WindowPointerLeaveRequest {
+  optional StreamControl control = 1;
+}
+message WindowPointerLeaveResponse {
+  // The window that the pointer left.
+  optional uint32 window_id = 1;
+}
+
+service SignalService {
+  rpc OutputConnect(stream OutputConnectRequest) returns (stream OutputConnectResponse);
+  rpc Layout(stream LayoutRequest) returns (stream LayoutResponse);
+  rpc WindowPointerEnter(stream WindowPointerEnterRequest) returns (stream WindowPointerEnterResponse);
+  rpc WindowPointerLeave(stream WindowPointerLeaveRequest) returns (stream WindowPointerLeaveResponse);
+}

api/rust/Cargo.toml

@@ -13,6 +13,7 @@ categories = ["api-bindings", "config"]
 pinnacle-api-defs = { workspace = true }
 pinnacle-api-macros = { path = "./pinnacle-api-macros" }
 tokio = { workspace = true, features = ["net"] }
+tokio-stream = { workspace = true }
 tonic = { workspace = true }
 tower = { version = "0.4.13", features = ["util"] }
 futures = "0.3.30"

api/rust/examples/default_config/main.rs

@@ -83,10 +83,10 @@ async fn main() {
 
     // Setup all monitors with tags "1" through "5"
     output.connect_for_all(move |op| {
-        let mut tags = tag.add(&op, tag_names);
+        let tags = tag.add(op, tag_names);
 
         // Be sure to set a tag to active or windows won't display
-        tags.next().unwrap().set_active(true);
+        tags.first().unwrap().set_active(true);
     });
 
     process.spawn_once([terminal]);

api/rust/src/input.rs

@@ -8,7 +8,7 @@
 //! methods for setting key- and mousebinds, changing xkeyboard settings, and more.
 //! View the struct's documentation for more information.
 
-use futures::{channel::mpsc::UnboundedSender, future::BoxFuture, FutureExt, StreamExt};
+use futures::{future::BoxFuture, FutureExt, StreamExt};
 use num_enum::TryFromPrimitive;
 use pinnacle_api_defs::pinnacle::input::{
     self,
@@ -19,6 +19,7 @@ use pinnacle_api_defs::pinnacle::input::{
         SetXkbConfigRequest,
     },
 };
+use tokio::sync::mpsc::UnboundedSender;
 use tonic::transport::Channel;
 use xkbcommon::xkb::Keysym;
 
@@ -162,7 +163,7 @@ impl Input {
         let modifiers = mods.into_iter().map(|modif| modif as i32).collect();
 
         self.fut_sender
-            .unbounded_send(
+            .send(
                 async move {
                     let mut stream = client
                         .set_keybind(SetKeybindRequest {
@@ -219,7 +220,7 @@ impl Input {
         let modifiers = mods.into_iter().map(|modif| modif as i32).collect();
 
         self.fut_sender
-            .unbounded_send(
+            .send(
                 async move {
                     let mut stream = client
                         .set_mousebind(SetMousebindRequest {

api/rust/src/lib.rs

@@ -82,7 +82,6 @@
 use std::sync::OnceLock;
 
 use futures::{
-    channel::mpsc::UnboundedReceiver,
     future::{BoxFuture, Either},
     stream::FuturesUnordered,
     Future, StreamExt,
@@ -91,7 +90,13 @@ use input::Input;
 use output::Output;
 use pinnacle::Pinnacle;
 use process::Process;
+use signal::SignalState;
 use tag::Tag;
+use tokio::sync::{
+    mpsc::{unbounded_channel, UnboundedReceiver},
+    RwLock,
+};
+use tokio_stream::wrappers::UnboundedReceiverStream;
 use tonic::transport::{Endpoint, Uri};
 use tower::service_fn;
 use window::Window;
@@ -100,6 +105,7 @@ pub mod input;
 pub mod output;
 pub mod pinnacle;
 pub mod process;
+pub mod signal;
 pub mod tag;
 pub mod util;
 pub mod window;
@@ -114,6 +120,7 @@ static WINDOW: OnceLock<Window> = OnceLock::new();
 static INPUT: OnceLock<Input> = OnceLock::new();
 static OUTPUT: OnceLock<Output> = OnceLock::new();
 static TAG: OnceLock<Tag> = OnceLock::new();
+static SIGNAL: OnceLock<RwLock<SignalState>> = OnceLock::new();
 
 /// A struct containing static references to all of the configuration structs.
 #[derive(Debug, Clone, Copy)]
@@ -147,16 +154,21 @@ pub async fn connect(
         }))
         .await?;
 
-    let (fut_sender, fut_recv) = futures::channel::mpsc::unbounded::<BoxFuture<()>>();
-
-    let output = Output::new(channel.clone(), fut_sender.clone());
+    let (fut_sender, fut_recv) = unbounded_channel::<BoxFuture<()>>();
 
     let pinnacle = PINNACLE.get_or_init(|| Pinnacle::new(channel.clone()));
     let process = PROCESS.get_or_init(|| Process::new(channel.clone(), fut_sender.clone()));
     let window = WINDOW.get_or_init(|| Window::new(channel.clone()));
     let input = INPUT.get_or_init(|| Input::new(channel.clone(), fut_sender.clone()));
-    let tag = TAG.get_or_init(|| Tag::new(channel.clone(), fut_sender.clone()));
-    let output = OUTPUT.get_or_init(|| output);
+    let tag = TAG.get_or_init(|| Tag::new(channel.clone()));
+    let output = OUTPUT.get_or_init(|| Output::new(channel.clone()));
+
+    SIGNAL
+        .set(RwLock::new(SignalState::new(
+            channel.clone(),
+            fut_sender.clone(),
+        )))
+        .map_err(|_| "failed to create SIGNAL")?;
 
     let modules = ApiModules {
         pinnacle,
@@ -177,9 +189,11 @@ pub async fn connect(
 ///
 /// This function is inserted at the end of your config through the [`config`] macro.
 /// You should use the macro instead of this function directly.
-pub async fn listen(mut fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>) {
+pub async fn listen(fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>) {
     let mut future_set = FuturesUnordered::<BoxFuture<()>>::new();
 
+    let mut fut_recv = UnboundedReceiverStream::new(fut_recv);
+
     loop {
         match futures::future::select(fut_recv.next(), future_set.next()).await {
             Either::Left((fut, _)) => {

api/rust/src/output.rs

@@ -9,39 +9,40 @@
 //! This module provides [`Output`], which allows you to get [`OutputHandle`]s for different
 //! connected monitors and set them up.
 
-use futures::{channel::mpsc::UnboundedSender, future::BoxFuture, FutureExt, StreamExt};
-use pinnacle_api_defs::pinnacle::{
-    output::{
+use futures::FutureExt;
+use pinnacle_api_defs::pinnacle::output::{
     self,
-        v0alpha1::{
-            output_service_client::OutputServiceClient, ConnectForAllRequest, SetLocationRequest,
-        },
-    },
-    tag::v0alpha1::tag_service_client::TagServiceClient,
+    v0alpha1::{output_service_client::OutputServiceClient, SetLocationRequest},
 };
 use tonic::transport::Channel;
 
-use crate::{block_on_tokio, tag::TagHandle, util::Batch};
+use crate::{
+    block_on_tokio,
+    signal::{OutputSignal, SignalHandle},
+    tag::TagHandle,
+    util::Batch,
+    SIGNAL, TAG,
+};
 
 /// A struct that allows you to get handles to connected outputs and set them up.
 ///
 /// See [`OutputHandle`] for more information.
 #[derive(Debug, Clone)]
 pub struct Output {
-    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
     output_client: OutputServiceClient<Channel>,
-    tag_client: TagServiceClient<Channel>,
 }
 
 impl Output {
-    pub(crate) fn new(
-        channel: Channel,
-        fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
-    ) -> Self {
+    pub(crate) fn new(channel: Channel) -> Self {
         Self {
             output_client: OutputServiceClient::new(channel.clone()),
-            tag_client: TagServiceClient::new(channel),
-            fut_sender,
+        }
+    }
+
+    pub(crate) fn new_handle(&self, name: impl Into<String>) -> OutputHandle {
+        OutputHandle {
+            name: name.into(),
+            output_client: self.output_client.clone(),
         }
     }
 
@@ -52,14 +53,14 @@ impl Output {
     /// ```
     /// let outputs = output.get_all();
     /// ```
-    pub fn get_all(&self) -> impl Iterator<Item = OutputHandle> {
+    pub fn get_all(&self) -> Vec<OutputHandle> {
         block_on_tokio(self.get_all_async())
     }
 
     /// The async version of [`Output::get_all`].
-    pub async fn get_all_async(&self) -> impl Iterator<Item = OutputHandle> {
+    pub async fn get_all_async(&self) -> Vec<OutputHandle> {
         let mut client = self.output_client.clone();
-        let tag_client = self.tag_client.clone();
+
         client
             .get(output::v0alpha1::GetRequest {})
             .await
@@ -67,11 +68,8 @@ impl Output {
             .into_inner()
             .output_names
             .into_iter()
-            .map(move |name| OutputHandle {
-                output_client: client.clone(),
-                tag_client: tag_client.clone(),
-                name,
-            })
+            .map(move |name| self.new_handle(name))
+            .collect()
     }
 
     /// Get a handle to the output with the given name.
@@ -93,6 +91,7 @@ impl Output {
         let name: String = name.into();
         self.get_all_async()
             .await
+            .into_iter()
             .find(|output| output.name == name)
     }
 
@@ -107,6 +106,7 @@ impl Output {
     /// ```
     pub fn get_focused(&self) -> Option<OutputHandle> {
         self.get_all()
+            .into_iter()
             .find(|output| matches!(output.props().focused, Some(true)))
     }
 
@@ -137,41 +137,26 @@ impl Output {
     ///     tags.next().unwrap().set_active(true);
     /// });
     /// ```
-    pub fn connect_for_all(&self, mut for_all: impl FnMut(OutputHandle) + Send + 'static) {
+    pub fn connect_for_all(&self, mut for_all: impl FnMut(&OutputHandle) + Send + 'static) {
         for output in self.get_all() {
-            for_all(output);
+            for_all(&output);
         }
 
-        let mut client = self.output_client.clone();
-        let tag_client = self.tag_client.clone();
-
-        self.fut_sender
-            .unbounded_send(
-                async move {
-                    let mut stream = client
-                        .connect_for_all(ConnectForAllRequest {})
-                        .await
-                        .unwrap()
-                        .into_inner();
-
-                    while let Some(Ok(response)) = stream.next().await {
-                        let Some(output_name) = response.output_name else {
-                            continue;
-                        };
-
-                        let output = OutputHandle {
-                            output_client: client.clone(),
-                            tag_client: tag_client.clone(),
-                            name: output_name,
-                        };
-
-                        for_all(output);
-                        tokio::task::yield_now().await;
+        let mut signal_state = block_on_tokio(SIGNAL.get().expect("SIGNAL doesn't exist").write());
+        signal_state.output_connect.add_callback(Box::new(for_all));
     }
+
+    /// Connect to an output signal.
+    ///
+    /// The compositor will fire off signals that your config can listen for and act upon.
+    /// You can pass in an [`OutputSignal`] along with a callback and it will get run
+    /// with the necessary arguments every time a signal of that type is received.
+    pub fn connect_signal(&self, signal: OutputSignal) -> SignalHandle {
+        let mut signal_state = block_on_tokio(SIGNAL.get().expect("SIGNAL doesn't exist").write());
+
+        match signal {
+            OutputSignal::Connect(f) => signal_state.output_connect.add_callback(f),
         }
-                .boxed(),
-            )
-            .unwrap();
     }
 }
 
@@ -180,9 +165,8 @@ impl Output {
 /// This allows you to manipulate outputs and get their properties.
 #[derive(Clone, Debug)]
 pub struct OutputHandle {
-    pub(crate) output_client: OutputServiceClient<Channel>,
-    pub(crate) tag_client: TagServiceClient<Channel>,
     pub(crate) name: String,
+    output_client: OutputServiceClient<Channel>,
 }
 
 impl PartialEq for OutputHandle {
@@ -408,6 +392,8 @@ impl OutputHandle {
             .unwrap()
             .into_inner();
 
+        let tag = TAG.get().expect("TAG doesn't exist");
+
         OutputProperties {
             make: response.make,
             model: response.model,
@@ -422,11 +408,7 @@ impl OutputHandle {
             tags: response
                 .tag_ids
                 .into_iter()
-                .map(|id| TagHandle {
-                    tag_client: self.tag_client.clone(),
-                    output_client: self.output_client.clone(),
-                    id,
-                })
+                .map(|id| tag.new_handle(id))
                 .collect(),
         }
     }

api/rust/src/process.rs

@@ -7,10 +7,11 @@
 //! This module provides [`Process`], which allows you to spawn processes and set environment
 //! variables.
 
-use futures::{channel::mpsc::UnboundedSender, future::BoxFuture, FutureExt, StreamExt};
+use futures::{future::BoxFuture, FutureExt, StreamExt};
 use pinnacle_api_defs::pinnacle::process::v0alpha1::{
     process_service_client::ProcessServiceClient, SetEnvRequest, SpawnRequest,
 };
+use tokio::sync::mpsc::UnboundedSender;
 use tonic::transport::Channel;
 
 use crate::block_on_tokio;
@@ -133,7 +134,7 @@ impl Process {
         };
 
         self.fut_sender
-            .unbounded_send(
+            .send(
                 async move {
                     let mut stream = client.spawn(request).await.unwrap().into_inner();
                     let Some(mut callbacks) = callbacks else { return };

api/rust/src/signal.rs

@@ -0,0 +1,394 @@
+//! Compositor signals.
+//!
+//! Your config can connect to various compositor signals that allow you to, for example, do
+//! something when an output is connected or when the pointer enters a window.
+//!
+//! Some of the other modules have a `connect_signal` method that will allow you to pass in
+//! callbacks to run on each signal. Use them to connect to the signals defined here.
+
+use std::{
+    collections::{btree_map, BTreeMap},
+    sync::{
+        atomic::{AtomicU32, Ordering},
+        Arc,
+    },
+};
+
+use futures::{future::BoxFuture, pin_mut, FutureExt};
+use pinnacle_api_defs::pinnacle::signal::v0alpha1::{
+    signal_service_client::SignalServiceClient, SignalRequest, StreamControl,
+};
+use tokio::sync::{
+    mpsc::{unbounded_channel, UnboundedSender},
+    oneshot,
+};
+use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt};
+use tonic::{transport::Channel, Streaming};
+
+use crate::{
+    block_on_tokio, output::OutputHandle, tag::TagHandle, window::WindowHandle, OUTPUT, TAG, WINDOW,
+};
+
+pub(crate) trait Signal {
+    type Callback;
+}
+
+macro_rules! signals {
+    ( $(
+        $( #[$cfg_enum:meta] )* $enum:ident => {
+            $(
+                $( #[$cfg:meta] )* $name:ident = {
+                    enum_name = $renamed:ident,
+                    callback_type = $cb:ty,
+                    client_request = $req:ident,
+                    on_response = $on_resp:expr,
+                }
+            )*
+        }
+    )* ) => {$(
+        $(
+            $( #[$cfg] )*
+            pub(crate) struct $name;
+
+            impl $crate::signal::Signal for $name {
+                type Callback = $cb;
+            }
+
+            impl SignalData<$name> {
+                pub(crate) fn add_callback(&mut self, callback: <$name as Signal>::Callback) -> SignalHandle {
+                    if self.callback_count.load(::std::sync::atomic::Ordering::SeqCst) == 0 {
+                        self.connect()
+                    }
+
+                    let Some(callback_sender) = self.callback_sender.as_ref() else {
+                        unreachable!("signal should already be connected here");
+                    };
+
+                    let Some(remove_callback_sender) = self.remove_callback_sender.clone() else {
+                        unreachable!("signal should already be connected here");
+                    };
+
+                    callback_sender
+                        .send((self.current_id, callback))
+                        .expect("failed to send callback");
+
+                    let handle = SignalHandle::new(self.current_id, remove_callback_sender);
+
+                    self.current_id.0 += 1;
+
+                    handle
+                }
+
+                fn reset(&mut self) {
+                    self.callback_sender.take();
+                    self.dc_pinger.take();
+                    self.remove_callback_sender.take();
+                    self.callback_count.store(0, Ordering::SeqCst);
+                    self.current_id = SignalConnId::default();
+                }
+
+                fn connect(&mut self) {
+                    self.reset();
+
+                    let channels = connect_signal::<_, _, <$name as Signal>::Callback, _, _>(
+                        &self.fut_sender,
+                        self.callback_count.clone(),
+                        |out| {
+                            block_on_tokio(self.client.$req(out))
+                                .expect("failed to request signal connection")
+                                .into_inner()
+                        },
+                        $on_resp,
+                    );
+
+                    self.callback_sender.replace(channels.callback_sender);
+                    self.dc_pinger.replace(channels.dc_pinger);
+                    self.remove_callback_sender
+                        .replace(channels.remove_callback_sender);
+                }
+            }
+        )*
+
+        $( #[$cfg_enum] )*
+        pub enum $enum {
+            $( $( #[$cfg] )* $renamed($cb),)*
+        }
+    )*};
+}
+
+signals! {
+    /// Signals relating to tag events.
+    TagSignal => {
+        /// The compositor requested that the given windows be laid out.
+        ///
+        /// Callbacks receive the tag that is being laid out and the windows being laid out.
+        ///
+        /// Note: if multiple tags are active, only the first will be received, but all windows on those
+        /// active tags will be received.
+        Layout = {
+            enum_name = Layout,
+            callback_type = LayoutFn,
+            client_request = layout,
+            on_response = |response, callbacks| {
+                if let Some(tag_id) = response.tag_id {
+                    let tag = TAG.get().expect("TAG doesn't exist");
+                    let window = WINDOW.get().expect("WINDOW doesn't exist");
+                    let tag = tag.new_handle(tag_id);
+
+                    let windows = response
+                        .window_ids
+                        .into_iter()
+                        .map(|id| window.new_handle(id))
+                        .collect::<Vec<_>>();
+
+                    for callback in callbacks {
+                        callback(&tag, windows.as_slice());
+                    }
+                }
+            },
+        }
+    }
+    /// Signals relating to output events.
+    OutputSignal => {
+        /// An output was connected.
+        ///
+        /// Callbacks receive the newly connected output.
+        ///
+        /// FIXME: This will not run on outputs that have been previously connected.
+        /// |      Tell the dev to fix this in the compositor.
+        OutputConnect = {
+            enum_name = Connect,
+            callback_type = SingleOutputFn,
+            client_request = output_connect,
+            on_response = |response, callbacks| {
+                if let Some(output_name) = response.output_name {
+                    let output = OUTPUT.get().expect("OUTPUT doesn't exist");
+                    let handle = output.new_handle(output_name);
+
+                    for callback in callbacks {
+                        callback(&handle);
+                    }
+                }
+            },
+        }
+    }
+    /// Signals relating to window events.
+    WindowSignal => {
+        /// The pointer entered a window.
+        ///
+        /// Callbacks receive the window the pointer entered.
+        WindowPointerEnter = {
+            enum_name = PointerEnter,
+            callback_type = SingleWindowFn,
+            client_request = window_pointer_enter,
+            on_response = |response, callbacks| {
+                if let Some(window_id) = response.window_id {
+                    let window = WINDOW.get().expect("WINDOW doesn't exist");
+                    let handle = window.new_handle(window_id);
+
+                    for callback in callbacks {
+                        callback(&handle);
+                    }
+                }
+            },
+        }
+        /// The pointer left a window.
+        ///
+        /// Callbacks receive the window the pointer left.
+        WindowPointerLeave = {
+            enum_name = PointerLeave,
+            callback_type = SingleWindowFn,
+            client_request = window_pointer_leave,
+            on_response = |response, callbacks| {
+                if let Some(window_id) = response.window_id {
+                    let window = WINDOW.get().expect("WINDOW doesn't exist");
+                    let handle = window.new_handle(window_id);
+
+                    for callback in callbacks {
+                        callback(&handle);
+                    }
+                }
+            },
+        }
+    }
+}
+
+pub(crate) type LayoutFn = Box<dyn FnMut(&TagHandle, &[WindowHandle]) + Send + 'static>;
+pub(crate) type SingleOutputFn = Box<dyn FnMut(&OutputHandle) + Send + 'static>;
+pub(crate) type SingleWindowFn = Box<dyn FnMut(&WindowHandle) + Send + 'static>;
+
+pub(crate) struct SignalState {
+    pub(crate) layout: SignalData<Layout>,
+    pub(crate) output_connect: SignalData<OutputConnect>,
+    pub(crate) window_pointer_enter: SignalData<WindowPointerEnter>,
+    pub(crate) window_pointer_leave: SignalData<WindowPointerLeave>,
+}
+
+impl SignalState {
+    pub(crate) fn new(
+        channel: Channel,
+        fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
+    ) -> Self {
+        let client = SignalServiceClient::new(channel);
+        Self {
+            layout: SignalData::new(client.clone(), fut_sender.clone()),
+            output_connect: SignalData::new(client.clone(), fut_sender.clone()),
+            window_pointer_enter: SignalData::new(client.clone(), fut_sender.clone()),
+            window_pointer_leave: SignalData::new(client.clone(), fut_sender.clone()),
+        }
+    }
+}
+
+#[derive(Default, Clone, Copy, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)]
+pub(crate) struct SignalConnId(pub(crate) u32);
+
+pub(crate) struct SignalData<S: Signal> {
+    client: SignalServiceClient<Channel>,
+    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
+    callback_sender: Option<UnboundedSender<(SignalConnId, S::Callback)>>,
+    remove_callback_sender: Option<UnboundedSender<SignalConnId>>,
+    dc_pinger: Option<oneshot::Sender<()>>,
+    current_id: SignalConnId,
+    callback_count: Arc<AtomicU32>,
+}
+
+impl<S: Signal> SignalData<S> {
+    fn new(
+        client: SignalServiceClient<Channel>,
+        fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
+    ) -> Self {
+        Self {
+            client,
+            fut_sender,
+            callback_sender: Default::default(),
+            remove_callback_sender: Default::default(),
+            dc_pinger: Default::default(),
+            current_id: Default::default(),
+            callback_count: Default::default(),
+        }
+    }
+}
+
+struct ConnectSignalChannels<F> {
+    callback_sender: UnboundedSender<(SignalConnId, F)>,
+    dc_pinger: oneshot::Sender<()>,
+    remove_callback_sender: UnboundedSender<SignalConnId>,
+}
+
+fn connect_signal<Req, Resp, F, T, O>(
+    fut_sender: &UnboundedSender<BoxFuture<'static, ()>>,
+    callback_count: Arc<AtomicU32>,
+    to_in_stream: T,
+    mut on_response: O,
+) -> ConnectSignalChannels<F>
+where
+    Req: SignalRequest + Send + 'static,
+    Resp: Send + 'static,
+    F: Send + 'static,
+    T: FnOnce(UnboundedReceiverStream<Req>) -> Streaming<Resp>,
+    O: FnMut(Resp, btree_map::ValuesMut<'_, SignalConnId, F>) + Send + 'static,
+{
+    let (control_sender, recv) = unbounded_channel::<Req>();
+    let out_stream = UnboundedReceiverStream::new(recv);
+
+    let mut in_stream = to_in_stream(out_stream);
+
+    let (callback_sender, mut callback_recv) = unbounded_channel::<(SignalConnId, F)>();
+    let (remove_callback_sender, mut remove_callback_recv) = unbounded_channel::<SignalConnId>();
+    let (dc_pinger, mut dc_ping_recv) = oneshot::channel::<()>();
+
+    let signal_future = async move {
+        let mut callbacks = BTreeMap::<SignalConnId, F>::new();
+
+        control_sender
+            .send(Req::from_control(StreamControl::Ready))
+            .expect("send failed");
+
+        loop {
+            let in_stream_next = in_stream.next().fuse();
+            pin_mut!(in_stream_next);
+            let callback_recv_recv = callback_recv.recv().fuse();
+            pin_mut!(callback_recv_recv);
+            let remove_callback_recv_recv = remove_callback_recv.recv().fuse();
+            pin_mut!(remove_callback_recv_recv);
+            let mut dc_ping_recv_fuse = (&mut dc_ping_recv).fuse();
+
+            futures::select! {
+                response = in_stream_next => {
+                    let Some(response) = response else { continue };
+
+                    match response {
+                        Ok(response) => {
+                            on_response(response, callbacks.values_mut());
+
+                            control_sender
+                                .send(Req::from_control(StreamControl::Ready))
+                                .expect("send failed");
+
+                            tokio::task::yield_now().await;
+                        }
+                        Err(status) => eprintln!("Error in recv: {status}"),
+                    }
+                }
+                callback = callback_recv_recv => {
+                    if let Some((id, callback)) = callback {
+                        callbacks.insert(id, callback);
+                        callback_count.fetch_add(1, Ordering::SeqCst);
+                    }
+                }
+                remove = remove_callback_recv_recv => {
+                    if let Some(id) = remove {
+                        if callbacks.remove(&id).is_some() {
+                            assert!(callback_count.fetch_sub(1, Ordering::SeqCst) > 0);
+                        }
+                        if callbacks.is_empty() {
+                            assert!(callback_count.load(Ordering::SeqCst) == 0);
+                            control_sender.send(Req::from_control(StreamControl::Disconnect)).expect("send failed");
+                            break;
+                        }
+                    }
+                }
+                _dc = dc_ping_recv_fuse => {
+                    println!("dc");
+                    control_sender.send(Req::from_control(StreamControl::Disconnect)).expect("send failed");
+                    break;
+                }
+            }
+        }
+    };
+
+    fut_sender.send(signal_future.boxed()).expect("send failed");
+
+    ConnectSignalChannels {
+        callback_sender,
+        dc_pinger,
+        remove_callback_sender,
+    }
+}
+
+/// A handle that can be used to disconnect from a signal connection.
+///
+/// This will remove the connected callback.
+pub struct SignalHandle {
+    id: SignalConnId,
+    remove_callback_sender: UnboundedSender<SignalConnId>,
+}
+
+impl SignalHandle {
+    pub(crate) fn new(
+        id: SignalConnId,
+        remove_callback_sender: UnboundedSender<SignalConnId>,
+    ) -> Self {
+        Self {
+            id,
+            remove_callback_sender,
+        }
+    }
+
+    /// Disconnect this signal connection.
+    pub fn disconnect(self) {
+        self.remove_callback_sender
+            .send(self.id)
+            .expect("failed to disconnect signal");
+    }
+}

api/rust/src/tag.rs

@@ -34,45 +34,42 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use futures::{channel::mpsc::UnboundedSender, future::BoxFuture, FutureExt};
+use futures::FutureExt;
 use num_enum::TryFromPrimitive;
-use pinnacle_api_defs::pinnacle::{
-    output::v0alpha1::output_service_client::OutputServiceClient,
-    tag::{
+use pinnacle_api_defs::pinnacle::tag::{
     self,
     v0alpha1::{
         tag_service_client::TagServiceClient, AddRequest, RemoveRequest, SetActiveRequest,
         SetLayoutRequest, SwitchToRequest,
     },
-    },
 };
 use tonic::transport::Channel;
 
 use crate::{
     block_on_tokio,
-    output::{Output, OutputHandle},
+    output::OutputHandle,
+    signal::{SignalHandle, TagSignal},
     util::Batch,
+    OUTPUT, SIGNAL,
 };
 
 /// A struct that allows you to add and remove tags and get [`TagHandle`]s.
 #[derive(Clone, Debug)]
 pub struct Tag {
-    channel: Channel,
-    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
     tag_client: TagServiceClient<Channel>,
-    output_client: OutputServiceClient<Channel>,
 }
 
 impl Tag {
-    pub(crate) fn new(
-        channel: Channel,
-        fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
-    ) -> Self {
+    pub(crate) fn new(channel: Channel) -> Self {
         Self {
             tag_client: TagServiceClient::new(channel.clone()),
-            output_client: OutputServiceClient::new(channel.clone()),
-            channel,
-            fut_sender,
+        }
+    }
+
+    pub(crate) fn new_handle(&self, id: u32) -> TagHandle {
+        TagHandle {
+            id,
+            tag_client: self.tag_client.clone(),
         }
     }
 
@@ -93,7 +90,7 @@ impl Tag {
         &self,
         output: &OutputHandle,
         tag_names: impl IntoIterator<Item = impl Into<String>>,
-    ) -> impl Iterator<Item = TagHandle> {
+    ) -> Vec<TagHandle> {
         block_on_tokio(self.add_async(output, tag_names))
     }
 
@@ -102,9 +99,8 @@ impl Tag {
         &self,
         output: &OutputHandle,
         tag_names: impl IntoIterator<Item = impl Into<String>>,
-    ) -> impl Iterator<Item = TagHandle> {
+    ) -> Vec<TagHandle> {
         let mut client = self.tag_client.clone();
-        let output_client = self.output_client.clone();
 
         let tag_names = tag_names.into_iter().map(Into::into).collect();
 
@@ -117,11 +113,11 @@ impl Tag {
             .unwrap()
             .into_inner();
 
-        response.tag_ids.into_iter().map(move |id| TagHandle {
-            tag_client: client.clone(),
-            output_client: output_client.clone(),
-            id,
-        })
+        response
+            .tag_ids
+            .into_iter()
+            .map(move |id| self.new_handle(id))
+            .collect()
     }
 
     /// Get handles to all tags across all outputs.
@@ -131,14 +127,13 @@ impl Tag {
     /// ```
     /// let all_tags = tag.get_all();
     /// ```
-    pub fn get_all(&self) -> impl Iterator<Item = TagHandle> {
+    pub fn get_all(&self) -> Vec<TagHandle> {
         block_on_tokio(self.get_all_async())
     }
 
     /// The async version of [`Tag::get_all`].
-    pub async fn get_all_async(&self) -> impl Iterator<Item = TagHandle> {
+    pub async fn get_all_async(&self) -> Vec<TagHandle> {
         let mut client = self.tag_client.clone();
-        let output_client = self.output_client.clone();
 
         let response = client
             .get(tag::v0alpha1::GetRequest {})
@@ -146,11 +141,11 @@ impl Tag {
             .unwrap()
             .into_inner();
 
-        response.tag_ids.into_iter().map(move |id| TagHandle {
-            tag_client: client.clone(),
-            output_client: output_client.clone(),
-            id,
-        })
+        response
+            .tag_ids
+            .into_iter()
+            .map(move |id| self.new_handle(id))
+            .collect()
     }
 
     /// Get a handle to the first tag with the given name on the focused output.
@@ -170,7 +165,7 @@ impl Tag {
     /// The async version of [`Tag::get`].
     pub async fn get_async(&self, name: impl Into<String>) -> Option<TagHandle> {
         let name = name.into();
-        let output_module = Output::new(self.channel.clone(), self.fut_sender.clone());
+        let output_module = OUTPUT.get().expect("OUTPUT doesn't exist");
         let focused_output = output_module.get_focused();
 
         if let Some(output) = focused_output {
@@ -280,7 +275,7 @@ impl Tag {
         let layouts_clone = layouts.clone();
         let len = layouts.len();
 
-        let output_module = Output::new(self.channel.clone(), self.fut_sender.clone());
+        let output_module = OUTPUT.get().expect("OUTPUT doesn't exist");
         let output_module_clone = output_module.clone();
 
         let next = move |output: Option<&OutputHandle>| {
@@ -343,6 +338,19 @@ impl Tag {
             next: Box::new(next),
         }
     }
+
+    /// Connect to a tag signal.
+    ///
+    /// The compositor will fire off signals that your config can listen for and act upon.
+    /// You can pass in a [`TagSignal`] along with a callback and it will get run
+    /// with the necessary arguments every time a signal of that type is received.
+    pub fn connect_signal(&self, signal: TagSignal) -> SignalHandle {
+        let mut signal_state = block_on_tokio(SIGNAL.get().expect("SIGNAL doesn't exist").write());
+
+        match signal {
+            TagSignal::Layout(layout_fn) => signal_state.layout.add_callback(layout_fn),
+        }
+    }
 }
 
 /// A layout cycler that keeps track of tags and their layouts and provides functions to cycle
@@ -361,8 +369,7 @@ pub struct LayoutCycler {
 #[derive(Debug, Clone)]
 pub struct TagHandle {
     pub(crate) id: u32,
-    pub(crate) tag_client: TagServiceClient<Channel>,
-    pub(crate) output_client: OutputServiceClient<Channel>,
+    tag_client: TagServiceClient<Channel>,
 }
 
 impl PartialEq for TagHandle {
@@ -543,7 +550,6 @@ impl TagHandle {
     /// The async version of [`TagHandle::props`].
     pub async fn props_async(&self) -> TagProperties {
         let mut client = self.tag_client.clone();
-        let output_client = self.output_client.clone();
 
         let response = client
             .get_properties(tag::v0alpha1::GetPropertiesRequest {
@@ -553,14 +559,12 @@ impl TagHandle {
             .unwrap()
             .into_inner();
 
+        let output = OUTPUT.get().expect("OUTPUT doesn't exist");
+
         TagProperties {
             active: response.active,
             name: response.name,
-            output: response.output_name.map(|name| OutputHandle {
-                output_client,
-                tag_client: client,
-                name,
-            }),
+            output: response.output_name.map(|name| output.new_handle(name)),
         }
     }
 

api/rust/src/window.rs

@@ -15,8 +15,6 @@
 use futures::FutureExt;
 use num_enum::TryFromPrimitive;
 use pinnacle_api_defs::pinnacle::{
-    output::v0alpha1::output_service_client::OutputServiceClient,
-    tag::v0alpha1::tag_service_client::TagServiceClient,
     window::v0alpha1::{
         window_service_client::WindowServiceClient, AddWindowRuleRequest, CloseRequest,
         MoveToTagRequest, SetTagRequest,
@@ -34,8 +32,10 @@ use tonic::transport::Channel;
 use crate::{
     block_on_tokio,
     input::MouseButton,
+    signal::{SignalHandle, WindowSignal},
     tag::TagHandle,
     util::{Batch, Geometry},
+    SIGNAL, TAG,
 };
 
 use self::rules::{WindowRule, WindowRuleCondition};
@@ -48,16 +48,19 @@ pub mod rules;
 #[derive(Debug, Clone)]
 pub struct Window {
     window_client: WindowServiceClient<Channel>,
-    tag_client: TagServiceClient<Channel>,
-    output_client: OutputServiceClient<Channel>,
 }
 
 impl Window {
     pub(crate) fn new(channel: Channel) -> Self {
         Self {
             window_client: WindowServiceClient::new(channel.clone()),
-            tag_client: TagServiceClient::new(channel.clone()),
-            output_client: OutputServiceClient::new(channel),
+        }
+    }
+
+    pub(crate) fn new_handle(&self, id: u32) -> WindowHandle {
+        WindowHandle {
+            id,
+            window_client: self.window_client.clone(),
         }
     }
 
@@ -119,15 +122,13 @@ impl Window {
     /// ```
     /// let windows = window.get_all();
     /// ```
-    pub fn get_all(&self) -> impl Iterator<Item = WindowHandle> {
+    pub fn get_all(&self) -> Vec<WindowHandle> {
         block_on_tokio(self.get_all_async())
     }
 
     /// The async version of [`Window::get_all`].
-    pub async fn get_all_async(&self) -> impl Iterator<Item = WindowHandle> {
+    pub async fn get_all_async(&self) -> Vec<WindowHandle> {
         let mut client = self.window_client.clone();
-        let tag_client = self.tag_client.clone();
-        let output_client = self.output_client.clone();
         client
             .get(GetRequest {})
             .await
@@ -135,12 +136,8 @@ impl Window {
             .into_inner()
             .window_ids
             .into_iter()
-            .map(move |id| WindowHandle {
-                window_client: client.clone(),
-                id,
-                tag_client: tag_client.clone(),
-                output_client: output_client.clone(),
-            })
+            .map(move |id| self.new_handle(id))
+            .collect::<Vec<_>>()
     }
 
     /// Get the currently focused window.
@@ -167,7 +164,7 @@ impl Window {
     /// A window rule is a set of criteria that a window must open with.
     /// For it to apply, a [`WindowRuleCondition`] must evaluate to true for the window in question.
     ///
-    /// TODO:
+    /// See the [`rules`] module for more information.
     pub fn add_window_rule(&self, cond: WindowRuleCondition, rule: WindowRule) {
         let mut client = self.window_client.clone();
 
@@ -177,6 +174,20 @@ impl Window {
         }))
         .unwrap();
     }
+
+    /// Connect to a window signal.
+    ///
+    /// The compositor will fire off signals that your config can listen for and act upon.
+    /// You can pass in a [`WindowSignal`] along with a callback and it will get run
+    /// with the necessary arguments every time a signal of that type is received.
+    pub fn connect_signal(&self, signal: WindowSignal) -> SignalHandle {
+        let mut signal_state = block_on_tokio(SIGNAL.get().expect("SIGNAL doesn't exist").write());
+
+        match signal {
+            WindowSignal::PointerEnter(f) => signal_state.window_pointer_enter.add_callback(f),
+            WindowSignal::PointerLeave(f) => signal_state.window_pointer_leave.add_callback(f),
+        }
+    }
 }
 
 /// A handle to a window.
@@ -186,8 +197,6 @@ impl Window {
 pub struct WindowHandle {
     id: u32,
     window_client: WindowServiceClient<Channel>,
-    tag_client: TagServiceClient<Channel>,
-    output_client: OutputServiceClient<Channel>,
 }
 
 impl PartialEq for WindowHandle {
@@ -476,7 +485,6 @@ impl WindowHandle {
     /// The async version of [`props`][Self::props].
     pub async fn props_async(&self) -> WindowProperties {
         let mut client = self.window_client.clone();
-        let tag_client = self.tag_client.clone();
 
         let response = match client
             .get_properties(window::v0alpha1::GetPropertiesRequest {
@@ -504,6 +512,8 @@ impl WindowHandle {
             height: geo.height() as u32,
         });
 
+        let tag = TAG.get().expect("TAG doesn't exist");
+
         WindowProperties {
             geometry,
             class: response.class,
@@ -514,11 +524,7 @@ impl WindowHandle {
             tags: response
                 .tag_ids
                 .into_iter()
-                .map(|id| TagHandle {
-                    tag_client: tag_client.clone(),
-                    output_client: self.output_client.clone(),
-                    id,
-                })
+                .map(|id| tag.new_handle(id))
                 .collect(),
         }
     }

pinnacle-api-defs/build.rs

@@ -13,6 +13,7 @@ fn main() {
         formatcp!("../api/protocol/pinnacle/process/{VERSION}/process.proto"),
         formatcp!("../api/protocol/pinnacle/tag/{VERSION}/tag.proto"),
         formatcp!("../api/protocol/pinnacle/window/{VERSION}/window.proto"),
+        formatcp!("../api/protocol/pinnacle/signal/{VERSION}/signal.proto"),
     ];
 
     let descriptor_path = PathBuf::from(std::env::var("OUT_DIR").unwrap()).join("pinnacle.bin");

pinnacle-api-defs/src/lib.rs

@@ -33,6 +33,42 @@ pub mod pinnacle {
             tonic::include_proto!("pinnacle.process.v0alpha1");
         }
     }
+
+    pub mod signal {
+        pub mod v0alpha1 {
+            tonic::include_proto!("pinnacle.signal.v0alpha1");
+
+            pub trait SignalRequest {
+                fn from_control(control: StreamControl) -> Self;
+                fn control(&self) -> StreamControl;
+            }
+
+            macro_rules! impl_signal_request {
+                ( $( $request:ident ),* ) => {
+                    $(
+                        impl SignalRequest for $request {
+                            fn from_control(control: StreamControl) -> Self {
+                                $request {
+                                    control: Some(control as i32),
+                                }
+                            }
+
+                            fn control(&self) -> StreamControl {
+                                self.control()
+                            }
+                        }
+                    )*
+                };
+            }
+
+            impl_signal_request!(
+                OutputConnectRequest,
+                LayoutRequest,
+                WindowPointerEnterRequest,
+                WindowPointerLeaveRequest
+            );
+        }
+    }
 }
 
 pub const FILE_DESCRIPTOR_SET: &[u8] = tonic::include_file_descriptor_set!("pinnacle");

src/api.rs

@@ -1,3 +1,5 @@
+pub mod signal;
+
 use std::{ffi::OsString, num::NonZeroU32, pin::Pin, process::Stdio};
 
 use pinnacle_api_defs::pinnacle::{
@@ -10,9 +12,7 @@ use pinnacle_api_defs::pinnacle::{
     },
     output::{
         self,
-        v0alpha1::{
-            output_service_server, ConnectForAllRequest, ConnectForAllResponse, SetLocationRequest,
-        },
+        v0alpha1::{output_service_server, SetLocationRequest},
     },
     process::v0alpha1::{process_service_server, SetEnvRequest, SpawnRequest, SpawnResponse},
     tag::{
@@ -44,9 +44,10 @@ use sysinfo::ProcessRefreshKind;
 use tokio::{
     io::AsyncBufReadExt,
     sync::mpsc::{unbounded_channel, UnboundedSender},
+    task::JoinHandle,
 };
-use tokio_stream::Stream;
-use tonic::{Request, Response, Status};
+use tokio_stream::{Stream, StreamExt};
+use tonic::{Request, Response, Status, Streaming};
 
 use crate::{
     config::ConnectorSavedState,
@@ -122,6 +123,46 @@ where
     Ok(Response::new(Box::pin(receiver_stream)))
 }
 
+fn run_bidirectional_streaming<F1, F2, I, O>(
+    fn_sender: StateFnSender,
+    mut in_stream: Streaming<I>,
+    with_client_request: F1,
+    with_out_stream: F2,
+) -> Result<Response<ResponseStream<O>>, Status>
+where
+    F1: Fn(&mut State, Result<I, Status>) + Clone + Send + 'static,
+    F2: FnOnce(&mut State, UnboundedSender<Result<O, Status>>, JoinHandle<()>) + Send + 'static,
+    I: Send + 'static,
+    O: Send + 'static,
+{
+    let (sender, receiver) = unbounded_channel::<Result<O, Status>>();
+
+    let fn_sender_clone = fn_sender.clone();
+
+    let with_in_stream = async move {
+        while let Some(request) = in_stream.next().await {
+            let with_client_request = with_client_request.clone();
+            // TODO: handle error
+            let _ = fn_sender_clone.send(Box::new(move |state: &mut State| {
+                with_client_request(state, request);
+            }));
+        }
+    };
+
+    let join_handle = tokio::spawn(with_in_stream);
+
+    let with_out_stream = Box::new(|state: &mut State| {
+        with_out_stream(state, sender, join_handle);
+    });
+
+    fn_sender
+        .send(with_out_stream)
+        .map_err(|_| Status::internal("failed to execute request"))?;
+
+    let receiver_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(receiver);
+    Ok(Response::new(Box::pin(receiver_stream)))
+}
+
 pub struct PinnacleService {
     sender: StateFnSender,
 }
@@ -690,6 +731,15 @@ impl tag_service_server::TagService for TagService {
             state.update_windows(&output);
             state.update_focus(&output);
             state.schedule_render(&output);
+
+            state.signal_state.layout.signal(|buffer| {
+                buffer.push_back(
+                    pinnacle_api_defs::pinnacle::signal::v0alpha1::LayoutResponse {
+                        window_ids: vec![1, 2, 3],
+                        tag_id: Some(1),
+                    },
+                );
+            });
         })
         .await
     }
@@ -892,8 +942,6 @@ impl OutputService {
 
 #[tonic::async_trait]
 impl output_service_server::OutputService for OutputService {
-    type ConnectForAllStream = ResponseStream<ConnectForAllResponse>;
-
     async fn set_location(
         &self,
         request: Request<SetLocationRequest>,
@@ -945,18 +993,6 @@ impl output_service_server::OutputService for OutputService {
         .await
     }
 
-    // TODO: remove this and integrate it into a signal/event system
-    async fn connect_for_all(
-        &self,
-        _request: Request<ConnectForAllRequest>,
-    ) -> Result<Response<Self::ConnectForAllStream>, Status> {
-        tracing::trace!("OutputService.connect_for_all");
-
-        run_server_streaming(&self.sender, move |state, sender| {
-            state.config.output_callback_senders.push(sender);
-        })
-    }
-
     async fn get(
         &self,
         _request: Request<output::v0alpha1::GetRequest>,
@@ -1075,7 +1111,7 @@ impl window_service_server::WindowService for WindowService {
     async fn close(&self, request: Request<CloseRequest>) -> Result<Response<()>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1104,7 +1140,7 @@ impl window_service_server::WindowService for WindowService {
 
         tracing::info!(request = ?request);
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1155,7 +1191,7 @@ impl window_service_server::WindowService for WindowService {
     ) -> Result<Response<()>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1198,7 +1234,7 @@ impl window_service_server::WindowService for WindowService {
     ) -> Result<Response<()>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1241,7 +1277,7 @@ impl window_service_server::WindowService for WindowService {
     ) -> Result<Response<()>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1284,7 +1320,7 @@ impl window_service_server::WindowService for WindowService {
     ) -> Result<Response<()>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1312,7 +1348,7 @@ impl window_service_server::WindowService for WindowService {
     async fn set_tag(&self, request: Request<SetTagRequest>) -> Result<Response<()>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,
@@ -1468,12 +1504,7 @@ impl window_service_server::WindowService for WindowService {
             let window_ids = state
                 .windows
                 .iter()
-                .map(|win| {
-                    win.with_state(|state| match state.id {
-                        WindowId::None => unreachable!(),
-                        WindowId::Some(id) => id,
-                    })
-                })
+                .map(|win| win.with_state(|state| state.id.0))
                 .collect::<Vec<_>>();
 
             window::v0alpha1::GetResponse { window_ids }
@@ -1487,7 +1518,7 @@ impl window_service_server::WindowService for WindowService {
     ) -> Result<Response<window::v0alpha1::GetPropertiesResponse>, Status> {
         let request = request.into_inner();
 
-        let window_id = WindowId::Some(
+        let window_id = WindowId(
             request
                 .window_id
                 .ok_or_else(|| Status::invalid_argument("no window specified"))?,

src/api/signal.rs

@@ -0,0 +1,222 @@
+use std::collections::VecDeque;
+
+use pinnacle_api_defs::pinnacle::signal::v0alpha1::{
+    signal_service_server, LayoutRequest, LayoutResponse, OutputConnectRequest,
+    OutputConnectResponse, SignalRequest, StreamControl, WindowPointerEnterRequest,
+    WindowPointerEnterResponse, WindowPointerLeaveRequest, WindowPointerLeaveResponse,
+};
+use tokio::{sync::mpsc::UnboundedSender, task::JoinHandle};
+use tonic::{Request, Response, Status, Streaming};
+
+use crate::state::State;
+
+use super::{run_bidirectional_streaming, ResponseStream, StateFnSender};
+
+#[derive(Debug, Default)]
+pub struct SignalState {
+    pub output_connect: SignalData<OutputConnectResponse, VecDeque<OutputConnectResponse>>,
+    pub layout: SignalData<LayoutResponse, VecDeque<LayoutResponse>>,
+    pub window_pointer_enter:
+        SignalData<WindowPointerEnterResponse, VecDeque<WindowPointerEnterResponse>>,
+    pub window_pointer_leave:
+        SignalData<WindowPointerLeaveResponse, VecDeque<WindowPointerLeaveResponse>>,
+}
+
+impl SignalState {
+    pub fn clear(&mut self) {
+        self.output_connect.disconnect();
+        self.layout.disconnect();
+        self.window_pointer_enter.disconnect();
+        self.window_pointer_leave.disconnect();
+    }
+}
+
+#[derive(Debug, Default)]
+#[allow(private_bounds)]
+pub struct SignalData<T, B: SignalBuffer<T>> {
+    sender: Option<UnboundedSender<Result<T, Status>>>,
+    join_handle: Option<JoinHandle<()>>,
+    ready: bool,
+    buffer: B,
+}
+
+/// A trait that denotes different types of containers that can be used to buffer signals.
+trait SignalBuffer<T>: Default {
+    /// Get the next signal from this buffer.
+    fn next(&mut self) -> Option<T>;
+}
+
+impl<T> SignalBuffer<T> for VecDeque<T> {
+    fn next(&mut self) -> Option<T> {
+        self.pop_front()
+    }
+}
+
+impl<T> SignalBuffer<T> for Option<T> {
+    fn next(&mut self) -> Option<T> {
+        self.take()
+    }
+}
+
+#[allow(private_bounds)]
+impl<T, B: SignalBuffer<T>> SignalData<T, B> {
+    /// Attempt to send a signal.
+    ///
+    /// If the client is ready to accept more of this signal, it will be sent immediately.
+    /// Otherwise, the signal will remain stored in the underlying buffer until the client is ready.
+    ///
+    /// Use `with_buffer` to populate and manipulate the buffer with the data you want.
+    pub fn signal(&mut self, with_buffer: impl FnOnce(&mut B)) {
+        let Some(sender) = self.sender.as_ref() else {
+            return;
+        };
+
+        with_buffer(&mut self.buffer);
+
+        if self.ready {
+            if let Some(data) = self.buffer.next() {
+                sender.send(Ok(data)).expect("failed to send signal");
+                self.ready = false;
+            }
+        }
+    }
+
+    pub fn connect(
+        &mut self,
+        sender: UnboundedSender<Result<T, Status>>,
+        join_handle: JoinHandle<()>,
+    ) {
+        self.sender.replace(sender);
+        if let Some(handle) = self.join_handle.replace(join_handle) {
+            handle.abort();
+        }
+    }
+
+    fn disconnect(&mut self) {
+        self.sender.take();
+        if let Some(handle) = self.join_handle.take() {
+            handle.abort();
+        }
+        self.ready = false;
+        self.buffer = B::default();
+    }
+
+    /// Mark this signal as ready to send.
+    ///
+    /// If there are signals already in the buffer, they will be sent.
+    fn ready(&mut self) {
+        let Some(sender) = self.sender.as_ref() else {
+            return;
+        };
+
+        if let Some(data) = self.buffer.next() {
+            sender.send(Ok(data)).expect("failed to send signal");
+            self.ready = false;
+        } else {
+            self.ready = true;
+        }
+    }
+}
+
+fn start_signal_stream<I, O, B, F>(
+    sender: StateFnSender,
+    in_stream: Streaming<I>,
+    with_signal_buffer: F,
+) -> Result<Response<ResponseStream<O>>, Status>
+where
+    I: SignalRequest + std::fmt::Debug + Send + 'static,
+    O: Send + 'static,
+    B: SignalBuffer<O>,
+    F: Fn(&mut State) -> &mut SignalData<O, B> + Clone + Send + 'static,
+{
+    let with_signal_buffer_clone = with_signal_buffer.clone();
+
+    run_bidirectional_streaming(
+        sender,
+        in_stream,
+        move |state, request| {
+            let request = match request {
+                Ok(request) => request,
+                Err(status) => {
+                    tracing::error!("Error in output_connect signal in stream: {status}");
+                    return;
+                }
+            };
+
+            tracing::debug!("Got {request:?} from client stream");
+
+            let signal = with_signal_buffer(state);
+            match request.control() {
+                StreamControl::Ready => signal.ready(),
+                StreamControl::Disconnect => signal.disconnect(),
+                StreamControl::Unspecified => tracing::warn!("Received unspecified stream control"),
+            }
+        },
+        move |state, sender, join_handle| {
+            let signal = with_signal_buffer_clone(state);
+            signal.connect(sender, join_handle);
+        },
+    )
+}
+
+pub struct SignalService {
+    sender: StateFnSender,
+}
+
+impl SignalService {
+    pub fn new(sender: StateFnSender) -> Self {
+        Self { sender }
+    }
+}
+
+#[tonic::async_trait]
+impl signal_service_server::SignalService for SignalService {
+    type OutputConnectStream = ResponseStream<OutputConnectResponse>;
+    type LayoutStream = ResponseStream<LayoutResponse>;
+    type WindowPointerEnterStream = ResponseStream<WindowPointerEnterResponse>;
+    type WindowPointerLeaveStream = ResponseStream<WindowPointerLeaveResponse>;
+
+    async fn output_connect(
+        &self,
+        request: Request<Streaming<OutputConnectRequest>>,
+    ) -> Result<Response<Self::OutputConnectStream>, Status> {
+        let in_stream = request.into_inner();
+
+        start_signal_stream(self.sender.clone(), in_stream, |state| {
+            &mut state.signal_state.output_connect
+        })
+    }
+
+    async fn layout(
+        &self,
+        request: Request<Streaming<LayoutRequest>>,
+    ) -> Result<Response<Self::LayoutStream>, Status> {
+        let in_stream = request.into_inner();
+
+        start_signal_stream(self.sender.clone(), in_stream, |state| {
+            &mut state.signal_state.layout
+        })
+    }
+
+    async fn window_pointer_enter(
+        &self,
+        request: Request<Streaming<WindowPointerEnterRequest>>,
+    ) -> Result<Response<Self::WindowPointerEnterStream>, Status> {
+        let in_stream = request.into_inner();
+
+        start_signal_stream(self.sender.clone(), in_stream, |state| {
+            &mut state.signal_state.window_pointer_enter
+        })
+    }
+
+    async fn window_pointer_leave(
+        &self,
+        request: Request<Streaming<WindowPointerLeaveRequest>>,
+    ) -> Result<Response<Self::WindowPointerLeaveStream>, Status> {
+        let in_stream = request.into_inner();
+
+        start_signal_stream(self.sender.clone(), in_stream, |state| {
+            &mut state.signal_state.window_pointer_leave
+        })
+    }
+}

src/backend/udev.rs

@@ -8,7 +8,7 @@ use std::{
 };
 
 use anyhow::Context;
-use pinnacle_api_defs::pinnacle::output::v0alpha1::ConnectForAllResponse;
+use pinnacle_api_defs::pinnacle::signal::v0alpha1::OutputConnectResponse;
 use smithay::{
     backend::{
         allocator::{
@@ -983,12 +983,11 @@ impl State {
 
             output.with_state(|state| state.tags = tags.clone());
         } else {
-            // Run any output callbacks
-            for sender in self.config.output_callback_senders.iter() {
-                let _ = sender.send(Ok(ConnectForAllResponse {
+            self.signal_state.output_connect.signal(|buffer| {
+                buffer.push_back(OutputConnectResponse {
                     output_name: Some(output.name()),
-                }));
-            }
+                })
+            });
         }
     }
 

src/config.rs

@@ -1,6 +1,7 @@
 use crate::{
     api::{
-        InputService, OutputService, PinnacleService, ProcessService, TagService, WindowService,
+        signal::SignalService, InputService, OutputService, PinnacleService, ProcessService,
+        TagService, WindowService,
     },
     input::ModifierMask,
     output::OutputName,
@@ -16,8 +17,9 @@ use std::{
 use anyhow::Context;
 use pinnacle_api_defs::pinnacle::{
     input::v0alpha1::input_service_server::InputServiceServer,
-    output::v0alpha1::{output_service_server::OutputServiceServer, ConnectForAllResponse},
+    output::v0alpha1::output_service_server::OutputServiceServer,
     process::v0alpha1::process_service_server::ProcessServiceServer,
+    signal::v0alpha1::signal_service_server::SignalServiceServer,
     tag::v0alpha1::tag_service_server::TagServiceServer,
     v0alpha1::pinnacle_service_server::PinnacleServiceServer,
     window::v0alpha1::window_service_server::WindowServiceServer,
@@ -28,7 +30,7 @@ use smithay::{
     utils::{Logical, Point},
 };
 use sysinfo::ProcessRefreshKind;
-use tokio::{sync::mpsc::UnboundedSender, task::JoinHandle};
+use tokio::task::JoinHandle;
 use toml::Table;
 
 use xdg::BaseDirectories;
@@ -164,7 +166,6 @@ pub enum Key {
 pub struct Config {
     /// Window rules and conditions on when those rules should apply
     pub window_rules: Vec<(WindowRuleCondition, WindowRule)>,
-    pub output_callback_senders: Vec<UnboundedSender<Result<ConnectForAllResponse, tonic::Status>>>,
     /// Saved states when outputs are disconnected
     pub connector_saved_states: HashMap<OutputName, ConnectorSavedState>,
 
@@ -175,7 +176,6 @@ pub struct Config {
 impl Config {
     pub fn clear(&mut self, loop_handle: &LoopHandle<State>) {
         self.window_rules.clear();
-        self.output_callback_senders.clear();
         self.connector_saved_states.clear();
         if let Some(join_handle) = self.config_join_handle.take() {
             join_handle.abort();
@@ -265,6 +265,8 @@ impl State {
 
         self.config.clear(&self.loop_handle);
 
+        self.signal_state.clear();
+
         // Because the grpc server is implemented to only start once,
         // any updates to `socket_dir` won't be applied until restart.
         if self.grpc_server_join_handle.is_none() {
@@ -447,6 +449,7 @@ impl State {
         let tag_service = TagService::new(grpc_sender.clone());
         let output_service = OutputService::new(grpc_sender.clone());
         let window_service = WindowService::new(grpc_sender.clone());
+        let signal_service = SignalService::new(grpc_sender.clone());
 
         let refl_service = tonic_reflection::server::Builder::configure()
             .register_encoded_file_descriptor_set(pinnacle_api_defs::FILE_DESCRIPTOR_SET)
@@ -464,7 +467,8 @@ impl State {
             .add_service(ProcessServiceServer::new(process_service))
             .add_service(TagServiceServer::new(tag_service))
             .add_service(OutputServiceServer::new(output_service))
-            .add_service(WindowServiceServer::new(window_service));
+            .add_service(WindowServiceServer::new(window_service))
+            .add_service(SignalServiceServer::new(signal_service));
 
         match self.xdisplay.as_ref() {
             Some(_) => {

src/state.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 use crate::{
-    backend::Backend, config::Config, cursor::Cursor, focus::FocusState,
+    api::signal::SignalState, backend::Backend, config::Config, cursor::Cursor, focus::FocusState,
     grab::resize_grab::ResizeSurfaceState, window::WindowElement,
 };
 use anyhow::Context;
@@ -93,6 +93,8 @@ pub struct State {
     pub grpc_server_join_handle: Option<tokio::task::JoinHandle<()>>,
 
     pub xdg_base_dirs: BaseDirectories,
+
+    pub signal_state: SignalState,
 }
 
 impl State {
@@ -269,6 +271,8 @@ impl State {
 
             xdg_base_dirs: BaseDirectories::with_prefix("pinnacle")
                 .context("couldn't create xdg BaseDirectories")?,
+
+            signal_state: SignalState::default(),
         };
 
         Ok(state)

src/window.rs

@@ -4,6 +4,9 @@ pub mod rules;
 
 use std::{cell::RefCell, time::Duration};
 
+use pinnacle_api_defs::pinnacle::signal::v0alpha1::{
+    WindowPointerEnterResponse, WindowPointerLeaveResponse,
+};
 use smithay::{
     backend::input::KeyState,
     desktop::{
@@ -330,33 +333,40 @@ impl WindowElement {
 }
 
 impl PointerTarget<State> for WindowElement {
-    fn frame(&self, seat: &Seat<State>, data: &mut State) {
+    fn frame(&self, seat: &Seat<State>, state: &mut State) {
         match self {
-            WindowElement::Wayland(window) => window.frame(seat, data),
+            WindowElement::Wayland(window) => window.frame(seat, state),
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                surface.frame(seat, data)
+                surface.frame(seat, state)
             }
             _ => unreachable!(),
         }
     }
 
-    fn enter(&self, seat: &Seat<State>, data: &mut State, event: &MotionEvent) {
+    fn enter(&self, seat: &Seat<State>, state: &mut State, event: &MotionEvent) {
         // TODO: ssd
         match self {
-            WindowElement::Wayland(window) => PointerTarget::enter(window, seat, data, event),
+            WindowElement::Wayland(window) => PointerTarget::enter(window, seat, state, event),
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                PointerTarget::enter(surface, seat, data, event)
+                PointerTarget::enter(surface, seat, state, event)
             }
             _ => unreachable!(),
         }
+
+        let window_id = Some(self.with_state(|state| state.id.0));
+
+        state
+            .signal_state
+            .window_pointer_enter
+            .signal(|buffer| buffer.push_back(WindowPointerEnterResponse { window_id }));
     }
 
-    fn motion(&self, seat: &Seat<State>, data: &mut State, event: &MotionEvent) {
+    fn motion(&self, seat: &Seat<State>, state: &mut State, event: &MotionEvent) {
         // TODO: ssd
         match self {
-            WindowElement::Wayland(window) => PointerTarget::motion(window, seat, data, event),
+            WindowElement::Wayland(window) => PointerTarget::motion(window, seat, state, event),
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                PointerTarget::motion(surface, seat, data, event)
+                PointerTarget::motion(surface, seat, state, event)
             }
             _ => unreachable!(),
         }
@@ -365,16 +375,16 @@ impl PointerTarget<State> for WindowElement {
     fn relative_motion(
         &self,
         seat: &Seat<State>,
-        data: &mut State,
+        state: &mut State,
         event: &smithay::input::pointer::RelativeMotionEvent,
     ) {
         // TODO: ssd
         match self {
             WindowElement::Wayland(window) => {
-                PointerTarget::relative_motion(window, seat, data, event);
+                PointerTarget::relative_motion(window, seat, state, event);
             }
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                PointerTarget::relative_motion(surface, seat, data, event);
+                PointerTarget::relative_motion(surface, seat, state, event);
             }
             _ => unreachable!(),
         }
@@ -383,47 +393,54 @@ impl PointerTarget<State> for WindowElement {
     fn button(
         &self,
         seat: &Seat<State>,
-        data: &mut State,
+        state: &mut State,
         event: &smithay::input::pointer::ButtonEvent,
     ) {
         // TODO: ssd
         match self {
-            WindowElement::Wayland(window) => PointerTarget::button(window, seat, data, event),
+            WindowElement::Wayland(window) => PointerTarget::button(window, seat, state, event),
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                PointerTarget::button(surface, seat, data, event)
+                PointerTarget::button(surface, seat, state, event)
             }
             _ => unreachable!(),
         }
     }
 
-    fn axis(&self, seat: &Seat<State>, data: &mut State, frame: AxisFrame) {
+    fn axis(&self, seat: &Seat<State>, state: &mut State, frame: AxisFrame) {
         // TODO: ssd
         match self {
-            WindowElement::Wayland(window) => PointerTarget::axis(window, seat, data, frame),
+            WindowElement::Wayland(window) => PointerTarget::axis(window, seat, state, frame),
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                PointerTarget::axis(surface, seat, data, frame)
+                PointerTarget::axis(surface, seat, state, frame)
             }
             _ => unreachable!(),
         }
     }
 
-    fn leave(&self, seat: &Seat<State>, data: &mut State, serial: Serial, time: u32) {
+    fn leave(&self, seat: &Seat<State>, state: &mut State, serial: Serial, time: u32) {
         // TODO: ssd
         match self {
             WindowElement::Wayland(window) => {
-                PointerTarget::leave(window, seat, data, serial, time);
+                PointerTarget::leave(window, seat, state, serial, time);
             }
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                PointerTarget::leave(surface, seat, data, serial, time)
+                PointerTarget::leave(surface, seat, state, serial, time)
             }
             _ => unreachable!(),
         }
+
+        let window_id = Some(self.with_state(|state| state.id.0));
+
+        state
+            .signal_state
+            .window_pointer_leave
+            .signal(|buffer| buffer.push_back(WindowPointerLeaveResponse { window_id }));
     }
 
     fn gesture_swipe_begin(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GestureSwipeBeginEvent,
     ) {
         todo!()
@@ -432,7 +449,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_swipe_update(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GestureSwipeUpdateEvent,
     ) {
         todo!()
@@ -441,7 +458,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_swipe_end(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GestureSwipeEndEvent,
     ) {
         todo!()
@@ -450,7 +467,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_pinch_begin(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GesturePinchBeginEvent,
     ) {
         todo!()
@@ -459,7 +476,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_pinch_update(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GesturePinchUpdateEvent,
     ) {
         todo!()
@@ -468,7 +485,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_pinch_end(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GesturePinchEndEvent,
     ) {
         todo!()
@@ -477,7 +494,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_hold_begin(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GestureHoldBeginEvent,
     ) {
         todo!()
@@ -486,7 +503,7 @@ impl PointerTarget<State> for WindowElement {
     fn gesture_hold_end(
         &self,
         _seat: &Seat<State>,
-        _data: &mut State,
+        _state: &mut State,
         _event: &smithay::input::pointer::GestureHoldEndEvent,
     ) {
         todo!()
@@ -497,26 +514,26 @@ impl KeyboardTarget<State> for WindowElement {
     fn enter(
         &self,
         seat: &Seat<State>,
-        data: &mut State,
+        state: &mut State,
         keys: Vec<KeysymHandle<'_>>,
         serial: Serial,
     ) {
         match self {
             WindowElement::Wayland(window) => {
-                KeyboardTarget::enter(window, seat, data, keys, serial);
+                KeyboardTarget::enter(window, seat, state, keys, serial);
             }
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                KeyboardTarget::enter(surface, seat, data, keys, serial)
+                KeyboardTarget::enter(surface, seat, state, keys, serial)
             }
             _ => unreachable!(),
         }
     }
 
-    fn leave(&self, seat: &Seat<State>, data: &mut State, serial: Serial) {
+    fn leave(&self, seat: &Seat<State>, state: &mut State, serial: Serial) {
         match self {
-            WindowElement::Wayland(window) => KeyboardTarget::leave(window, seat, data, serial),
+            WindowElement::Wayland(window) => KeyboardTarget::leave(window, seat, state, serial),
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                KeyboardTarget::leave(surface, seat, data, serial)
+                KeyboardTarget::leave(surface, seat, state, serial)
             }
             _ => unreachable!(),
         }
@@ -525,18 +542,18 @@ impl KeyboardTarget<State> for WindowElement {
     fn key(
         &self,
         seat: &Seat<State>,
-        data: &mut State,
+        state: &mut State,
         key: KeysymHandle<'_>,
-        state: KeyState,
+        key_state: KeyState,
         serial: Serial,
         time: u32,
     ) {
         match self {
             WindowElement::Wayland(window) => {
-                KeyboardTarget::key(window, seat, data, key, state, serial, time);
+                KeyboardTarget::key(window, seat, state, key, key_state, serial, time);
             }
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                KeyboardTarget::key(surface, seat, data, key, state, serial, time);
+                KeyboardTarget::key(surface, seat, state, key, key_state, serial, time);
             }
             _ => unreachable!(),
         }
@@ -545,16 +562,16 @@ impl KeyboardTarget<State> for WindowElement {
     fn modifiers(
         &self,
         seat: &Seat<State>,
-        data: &mut State,
+        state: &mut State,
         modifiers: ModifiersState,
         serial: Serial,
     ) {
         match self {
             WindowElement::Wayland(window) => {
-                KeyboardTarget::modifiers(window, seat, data, modifiers, serial);
+                KeyboardTarget::modifiers(window, seat, state, modifiers, serial);
             }
             WindowElement::X11(surface) | WindowElement::X11OverrideRedirect(surface) => {
-                KeyboardTarget::modifiers(surface, seat, data, modifiers, serial);
+                KeyboardTarget::modifiers(surface, seat, state, modifiers, serial);
             }
             _ => unreachable!(),
         }

src/window/window_state.rs

@@ -17,20 +17,14 @@ use super::WindowElement;
 
 /// A unique identifier for each window.
 #[derive(Debug, Hash, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-pub enum WindowId {
-    /// A config API returned an invalid window. It should be using this variant.
-    None,
-    /// A valid window id.
-    #[serde(untagged)]
-    Some(u32),
-}
+pub struct WindowId(pub u32);
 
 static WINDOW_ID_COUNTER: AtomicU32 = AtomicU32::new(0);
 
 impl WindowId {
     /// Get the next available window id. This always starts at 0.
     pub fn next() -> Self {
-        Self::Some(WINDOW_ID_COUNTER.fetch_add(1, Ordering::Relaxed))
+        Self(WINDOW_ID_COUNTER.fetch_add(1, Ordering::Relaxed))
     }
 
     /// Get the window that has this WindowId.