Merge pull request #132 from pinnacle-comp/rust_api_tokio_and_futures

Get the futures to work

api/rust_grpc/Cargo.toml

@@ -7,10 +7,13 @@ edition = "2021"
 pinnacle-api-defs = { path = "../../pinnacle-api-defs" }
 pinnacle-api-macros = { path = "./pinnacle-api-macros" }
 tokio = { version = "1.35.1", features = ["macros", "rt-multi-thread", "net"] }
-tokio-stream = { version = "0.1.14", features = ["net"] }
+# tokio-stream = { version = "0.1.14", features = ["net"] }
+async-net = "2.0.0"
+async-compat = "0.2.3"
 tonic = "0.10.2"
 tower = { version = "0.4.13", features = ["util"] }
 futures = "0.3.30"
+# futures-lite = "2.2.0"
 num_enum = "0.7.2"
 xkbcommon = "0.7.0"
 

api/rust_grpc/examples/default_config.rs

@@ -1,18 +0,0 @@
-use pinnacle_api::{input::Mod, ApiModules};
-
-#[pinnacle_api::config(modules)]
-#[tokio::main]
-async fn main() {
-    let ApiModules {
-        pinnacle,
-        process,
-        window,
-        input,
-        output,
-        tag,
-    } = modules;
-
-    input.keybind([Mod::Shift], 'a', || {
-        process.spawn(["alacritty"]);
-    });
-}

api/rust_grpc/examples/default_config/main.rs

@@ -0,0 +1,103 @@
+use pinnacle_api::{
+    input::{Mod, MouseButton, MouseEdge},
+    ApiModules,
+};
+use xkbcommon::xkb::keysyms;
+
+#[pinnacle_api::config(modules)]
+#[tokio::main]
+async fn main() {
+    let ApiModules {
+        pinnacle,
+        process,
+        window,
+        input,
+        output,
+        tag,
+    } = modules;
+
+    let mod_key = Mod::Ctrl;
+
+    input.mousebind([mod_key], MouseButton::Left, MouseEdge::Press, || {
+        window.begin_move(MouseButton::Left);
+    });
+
+    input.mousebind([mod_key], MouseButton::Right, MouseEdge::Press, || {
+        window.begin_resize(MouseButton::Right);
+    });
+
+    // Keybinds
+
+    input.keybind([mod_key, Mod::Alt], 'q', || {
+        pinnacle.quit();
+    });
+
+    input.keybind([mod_key, Mod::Alt], 'c', || {
+        if let Some(window) = window.get_focused() {
+            window.close();
+        }
+    });
+
+    input.keybind([mod_key], keysyms::KEY_Return, || {
+        process.spawn(["alacritty"]);
+    });
+
+    input.keybind([mod_key, Mod::Alt], keysyms::KEY_space, || {
+        if let Some(window) = window.get_focused() {
+            window.toggle_floating();
+        }
+    });
+
+    input.keybind([mod_key], 'f', || {
+        if let Some(window) = window.get_focused() {
+            window.toggle_fullscreen();
+        }
+    });
+
+    input.keybind([mod_key], 'm', || {
+        if let Some(window) = window.get_focused() {
+            window.toggle_maximized();
+        }
+    });
+
+    // Tags
+
+    let tag_names = ["1", "2", "3", "4", "5"];
+
+    output.connect_for_all(move |op| {
+        let mut tags = tag.add(&op, tag_names);
+        tags.next().unwrap().set_active(true);
+    });
+
+    process.spawn_once(["alacritty"]);
+
+    for tag_name in tag_names {
+        input.keybind([mod_key], tag_name, move || {
+            if let Some(tg) = tag.get(tag_name, None) {
+                tg.switch_to();
+            }
+        });
+
+        input.keybind([mod_key, Mod::Shift], tag_name, move || {
+            if let Some(tg) = tag.get(tag_name, None) {
+                tg.toggle_active();
+            }
+        });
+
+        input.keybind([mod_key, Mod::Alt], tag_name, move || {
+            if let Some(tg) = tag.get(tag_name, None) {
+                if let Some(win) = window.get_focused() {
+                    win.move_to_tag(&tg);
+                }
+            }
+        });
+
+        input.keybind([mod_key, Mod::Shift, Mod::Alt], tag_name, move || {
+            if let Some(tg) = tag.get(tag_name, None) {
+                if let Some(win) = window.get_focused() {
+                    win.toggle_tag(&tg);
+                }
+            }
+        });
+    }
+}

api/rust_grpc/examples/default_config/metaconfig.toml

@@ -0,0 +1,46 @@
+# This metaconfig.toml file dictates what config Pinnacle will run.
+#
+# When running Pinnacle, the compositor will look in the following directories for a metaconfig.toml file,
+# in order from top to bottom:
+#     $PINNACLE_CONFIG_DIR
+#     $XDG_CONFIG_HOME/pinnacle/
+#     ~/.config/pinnacle/
+#
+# When Pinnacle finds a metaconfig.toml file, it will execute the command provided to `command`.
+# To use a Rust config, this should be changed to something like ["cargo", "run"].
+# 
+# Because configuration is done using an external process, if it ever crashes, you lose all of your keybinds.
+# The compositor will load the default config if that happens, but in the event that you don't have
+# the necessary dependencies for it to run, you may get softlocked.
+# In order prevent you from getting stuck in the compositor, you must define keybinds to reload your config
+# and kill Pinnacle.
+# 
+# More details on each setting can be found below.
+
+# The command Pinnacle will run on startup and when you reload your config.
+# Paths are relative to the directory the metaconfig.toml file is in.
+# This must be an array.
+command = ["cargo", "run", "--example", "default_config"]
+
+### Keybinds ###
+# Each keybind takes in a table with two fields: `modifiers` and `key`.
+# - `modifiers` can be one of "Ctrl", "Alt", "Shift", or "Super".
+# - `key` can be a string of any lowercase letter, number,
+#   "numN" where N is a number for numpad keys, or "esc"/"escape".
+# Support for any xkbcommon key is planned for a future update.
+
+# The keybind that will reload your config.
+reload_keybind = { modifiers = ["Ctrl", "Alt"], key = "r" }
+# The keybind that will kill Pinnacle.
+kill_keybind = { modifiers = ["Ctrl", "Alt", "Shift"], key = "escape" }
+
+### Socket directory ###
+# Pinnacle will open a Unix socket at `$XDG_RUNTIME_DIR` by default, falling back to `/tmp` if it doesn't exist.
+# If you want/need to change this, use the `socket_dir` setting set to the directory of your choosing.
+#
+# socket_dir = "/your/dir/here/"
+
+### Environment Variables ###
+# If you need to spawn your config with any environment variables, list them here.
+[envs]
+# key = "value"

api/rust_grpc/pinnacle-api-macros/src/lib.rs

@@ -20,11 +20,11 @@ pub fn config(
     quote! {
         #(#attrs)*
         #vis #sig {
-            let (#module_ident, __fut_receiver) = ::pinnacle_api::create_modules().unwrap();
+            let (#module_ident, __fut_receiver) = ::pinnacle_api::connect().await.unwrap();
 
             #(#stmts)*
 
-            ::pinnacle_api::listen(__fut_receiver);
+            ::pinnacle_api::listen(__fut_receiver).await;
         }
     }
     .into()

api/rust_grpc/src/input.rs

@@ -1,4 +1,6 @@
-use futures::executor::block_on;
+use futures::{
+    channel::mpsc::UnboundedSender, executor::block_on, future::BoxFuture, FutureExt, StreamExt,
+};
 use num_enum::TryFromPrimitive;
 use pinnacle_api_defs::pinnacle::input::{
     self,
@@ -8,14 +10,11 @@ use pinnacle_api_defs::pinnacle::input::{
         SetKeybindRequest, SetLibinputSettingRequest, SetMousebindRequest, SetRepeatRateRequest,
     },
 };
-use tokio_stream::StreamExt;
 use tonic::transport::Channel;
 use xkbcommon::xkb::Keysym;
 
 pub use pinnacle_api_defs::pinnacle::input::v0alpha1::SetXkbConfigRequest as XkbConfig;
 
-use crate::FutSender;
-
 use self::libinput::LibinputSetting;
 pub mod libinput;
 
@@ -48,13 +47,21 @@ pub enum MouseEdge {
 
 #[derive(Debug, Clone)]
 pub struct Input {
-    client: InputServiceClient<Channel>,
+    // client: InputServiceClient<Channel>,
-    fut_sender: FutSender,
+    channel: Channel,
+    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
 }
 
 impl Input {
-    pub fn new(client: InputServiceClient<Channel>, fut_sender: FutSender) -> Self {
+    pub fn new(channel: Channel, fut_sender: UnboundedSender<BoxFuture<'static, ()>>) -> Self {
-        Self { client, fut_sender }
+        Self {
+            channel,
+            fut_sender,
+        }
+    }
+
+    fn create_input_client(&self) -> InputServiceClient<Channel> {
+        InputServiceClient::new(self.channel.clone())
     }
 
     pub fn keybind(
@@ -63,27 +70,30 @@ impl Input {
         key: impl Key + Send + 'static,
         mut action: impl FnMut() + Send + 'static,
     ) {
-        let mut client = self.client.clone();
+        let mut client = self.create_input_client();
 
         let modifiers = mods.into_iter().map(|modif| modif as i32).collect();
 
         self.fut_sender
-            .send(Box::pin(async move {
+            .unbounded_send(
-                let mut stream = client
+                async move {
-                    .set_keybind(SetKeybindRequest {
+                    let mut stream = client
-                        modifiers,
+                        .set_keybind(SetKeybindRequest {
-                        key: Some(input::v0alpha1::set_keybind_request::Key::RawCode(
+                            modifiers,
-                            key.into_keysym().raw(),
+                            key: Some(input::v0alpha1::set_keybind_request::Key::RawCode(
-                        )),
+                                key.into_keysym().raw(),
-                    })
+                            )),
-                    .await
+                        })
-                    .unwrap()
+                        .await
-                    .into_inner();
+                        .unwrap()
+                        .into_inner();
 
-                while let Some(Ok(_response)) = stream.next().await {
+                    while let Some(Ok(_response)) = stream.next().await {
-                    action();
+                        action();
+                    }
                 }
-            }))
+                .boxed(),
+            )
             .unwrap();
     }
 
@@ -94,35 +104,40 @@ impl Input {
         edge: MouseEdge,
         mut action: impl FnMut() + 'static + Send,
     ) {
-        let mut client = self.client.clone();
+        let mut client = self.create_input_client();
 
         let modifiers = mods.into_iter().map(|modif| modif as i32).collect();
 
-        tokio::spawn(async move {
+        self.fut_sender
-            let mut stream = client
+            .unbounded_send(
-                .set_mousebind(SetMousebindRequest {
+                async move {
-                    modifiers,
+                    let mut stream = client
-                    button: Some(button as u32),
+                        .set_mousebind(SetMousebindRequest {
-                    edge: Some(edge as i32),
+                            modifiers,
-                })
+                            button: Some(button as u32),
-                .await
+                            edge: Some(edge as i32),
-                .unwrap()
+                        })
-                .into_inner();
+                        .await
+                        .unwrap()
+                        .into_inner();
 
-            while let Some(Ok(_response)) = stream.next().await {
+                    while let Some(Ok(_response)) = stream.next().await {
-                action();
+                        action();
-            }
+                    }
-        });
+                }
+                .boxed(),
+            )
+            .unwrap();
     }
 
     pub fn set_xkb_config(&self, xkb_config: XkbConfig) {
-        let mut client = self.client.clone();
+        let mut client = self.create_input_client();
 
         block_on(client.set_xkb_config(xkb_config)).unwrap();
     }
 
     pub fn set_repeat_rate(&self, rate: i32, delay: i32) {
-        let mut client = self.client.clone();
+        let mut client = self.create_input_client();
 
         block_on(client.set_repeat_rate(SetRepeatRateRequest {
             rate: Some(rate),
@@ -132,7 +147,7 @@ impl Input {
     }
 
     pub fn set_libinput_setting(&self, setting: LibinputSetting) {
-        let mut client = self.client.clone();
+        let mut client = self.create_input_client();
 
         let setting = match setting {
             LibinputSetting::AccelProfile(profile) => Setting::AccelProfile(profile as i32),

api/rust_grpc/src/lib.rs

@@ -2,23 +2,15 @@
 
 use std::sync::OnceLock;
 
-use futures::{executor::block_on, future::BoxFuture, stream::FuturesUnordered, StreamExt};
+use futures::{
+    channel::mpsc::UnboundedReceiver, future::BoxFuture, stream::FuturesUnordered, StreamExt,
+};
 use input::Input;
 use output::Output;
 use pinnacle::Pinnacle;
-use pinnacle_api_defs::pinnacle::{
-    input::v0alpha1::input_service_client::InputServiceClient,
-    output::v0alpha1::output_service_client::OutputServiceClient,
-    process::v0alpha1::process_service_client::ProcessServiceClient,
-    tag::v0alpha1::tag_service_client::TagServiceClient,
-    v0alpha1::pinnacle_service_client::PinnacleServiceClient,
-    window::v0alpha1::window_service_client::WindowServiceClient,
-};
 use process::Process;
 use tag::Tag;
-use tokio::{net::UnixStream, sync::mpsc::UnboundedSender};
+use tonic::transport::{Endpoint, Uri};
-use tokio_stream::wrappers::UnboundedReceiverStream;
-use tonic::transport::{Channel, Endpoint, Uri};
 use tower::service_fn;
 use window::Window;
 
@@ -31,6 +23,7 @@ pub mod util;
 pub mod window;
 
 pub use pinnacle_api_macros::config;
+pub use xkbcommon;
 
 static PINNACLE: OnceLock<Pinnacle> = OnceLock::new();
 static PROCESS: OnceLock<Process> = OnceLock::new();
@@ -39,30 +32,37 @@ static INPUT: OnceLock<Input> = OnceLock::new();
 static OUTPUT: OnceLock<Output> = OnceLock::new();
 static TAG: OnceLock<Tag> = OnceLock::new();
 
-pub(crate) type FutSender = UnboundedSender<BoxFuture<'static, ()>>;
+#[derive(Debug, Clone)]
+pub struct ApiModules {
+    pub pinnacle: &'static Pinnacle,
+    pub process: &'static Process,
+    pub window: &'static Window,
+    pub input: &'static Input,
+    pub output: &'static Output,
+    pub tag: &'static Tag,
+}
 
-pub fn create_modules(
+pub async fn connect(
-) -> Result<(ApiModules, UnboundedReceiverStream<BoxFuture<'static, ()>>), Box<dyn std::error::Error>>
+) -> Result<(ApiModules, UnboundedReceiver<BoxFuture<'static, ()>>), Box<dyn std::error::Error>> {
-{
+    let channel = Endpoint::try_from("http://[::]:50051")? // port doesn't matter, we use a unix socket
-    let channel = connect()?;
+        .connect_with_connector(service_fn(|_: Uri| {
+            tokio::net::UnixStream::connect(
+                std::env::var("PINNACLE_GRPC_SOCKET")
+                    .expect("PINNACLE_GRPC_SOCKET was not set; is Pinnacle running?"),
+            )
+        }))
+        .await?;
 
-    let pinnacle_client = PinnacleServiceClient::new(channel.clone());
+    let (fut_sender, fut_recv) = futures::channel::mpsc::unbounded::<BoxFuture<()>>();
-    let window_client = WindowServiceClient::new(channel.clone());
-    let input_client = InputServiceClient::new(channel.clone());
-    let output_client = OutputServiceClient::new(channel.clone());
-    let tag_client = TagServiceClient::new(channel.clone());
-    let process_client = ProcessServiceClient::new(channel.clone());
 
-    let (fut_sender, fut_receiver) = tokio::sync::mpsc::unbounded_channel::<BoxFuture<()>>();
+    let output = Output::new(channel.clone(), fut_sender.clone());
 
-    let fut_receiver = UnboundedReceiverStream::new(fut_receiver);
+    let pinnacle = PINNACLE.get_or_init(|| Pinnacle::new(channel.clone()));
-
+    let process = PROCESS.get_or_init(|| Process::new(channel.clone(), fut_sender.clone()));
-    let pinnacle = PINNACLE.get_or_init(|| Pinnacle::new(pinnacle_client));
+    let window = WINDOW.get_or_init(|| Window::new(channel.clone()));
-    let process = PROCESS.get_or_init(|| Process::new(process_client, fut_sender.clone()));
+    let input = INPUT.get_or_init(|| Input::new(channel.clone(), fut_sender.clone()));
-    let window = WINDOW.get_or_init(|| Window::new(window_client, tag_client.clone()));
+    let tag = TAG.get_or_init(|| Tag::new(channel.clone(), fut_sender.clone()));
-    let input = INPUT.get_or_init(|| Input::new(input_client, fut_sender.clone()));
+    let output = OUTPUT.get_or_init(|| output);
-    let output = OUTPUT.get_or_init(|| Output::new(output_client, tag_client.clone()));
-    let tag = TAG.get_or_init(|| Tag::new(tag_client));
 
     let modules = ApiModules {
         pinnacle,
@@ -73,63 +73,36 @@ pub fn create_modules(
         tag,
     };
 
-    Ok((modules, fut_receiver))
+    Ok((modules, fut_recv))
 }
 
-pub fn listen(
+pub async fn listen(
-    fut_receiver: UnboundedReceiverStream<BoxFuture<()>>,
+    fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>, // api_modules: ApiModules<'a>,
-    // api_modules: ApiModules<'a>,
 ) {
     let mut future_set = FuturesUnordered::<
         BoxFuture<(
             Option<BoxFuture<()>>,
-            Option<UnboundedReceiverStream<BoxFuture<()>>>,
+            Option<UnboundedReceiver<BoxFuture<()>>>,
         )>,
     >::new();
 
     future_set.push(Box::pin(async move {
-        let (fut, stream) = fut_receiver.into_future().await;
+        let (fut, stream) = fut_recv.into_future().await;
         (fut, Some(stream))
     }));
 
-    block_on(async move {
+    while let Some((fut, stream)) = future_set.next().await {
-        while let Some((fut, stream)) = future_set.next().await {
+        if let Some(fut) = fut {
-            if let Some(fut) = fut {
+            future_set.push(Box::pin(async move {
-                future_set.push(Box::pin(async move {
+                fut.await;
-                    fut.await;
+                (None, None)
-                    (None, None)
+            }));
-                }));
-            }
-            if let Some(stream) = stream {
-                future_set.push(Box::pin(async move {
-                    let (fut, stream) = stream.into_future().await;
-                    (fut, Some(stream))
-                }))
-            }
         }
-    });
+        if let Some(stream) = stream {
-}
+            future_set.push(Box::pin(async move {
-
+                let (fut, stream) = stream.into_future().await;
-// #[derive(Debug, Clone)]
+                (fut, Some(stream))
-pub struct ApiModules {
-    pub pinnacle: &'static Pinnacle,
-    pub process: &'static Process,
-    pub window: &'static Window,
-    pub input: &'static Input,
-    pub output: &'static Output,
-    pub tag: &'static Tag,
-}
-
-pub fn connect() -> Result<Channel, Box<dyn std::error::Error>> {
-    block_on(async {
-        Endpoint::try_from("http://[::]:50051")? // port doesn't matter, we use a unix socket
-            .connect_with_connector(service_fn(|_: Uri| {
-                UnixStream::connect(
-                    std::env::var("PINNACLE_GRPC_SOCKET")
-                        .expect("PINNACLE_GRPC_SOCKET was not set; is Pinnacle running?"),
-                )
             }))
-            .await
+        }
-    })
+    }
-    .map_err(|err| err.into())
 }

api/rust_grpc/src/output.rs

@@ -1,4 +1,6 @@
-use futures::executor::block_on;
+use futures::{
+    channel::mpsc::UnboundedSender, executor::block_on, future::BoxFuture, FutureExt, StreamExt,
+};
 use pinnacle_api_defs::pinnacle::{
     output::{
         self,
@@ -8,28 +10,37 @@ use pinnacle_api_defs::pinnacle::{
     },
     tag::v0alpha1::tag_service_client::TagServiceClient,
 };
-use tokio_stream::StreamExt;
 use tonic::transport::Channel;
 
 use crate::tag::TagHandle;
 
 #[derive(Debug, Clone)]
 pub struct Output {
-    client: OutputServiceClient<Channel>,
+    // client: OutputServiceClient<Channel>,
-    tag_client: TagServiceClient<Channel>,
+    // tag_client: TagServiceClient<Channel>,
+    channel: Channel,
+    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
 }
 
 impl Output {
-    pub fn new(
+    pub fn new(channel: Channel, fut_sender: UnboundedSender<BoxFuture<'static, ()>>) -> Self {
-        client: OutputServiceClient<Channel>,
+        Self {
-        tag_client: TagServiceClient<Channel>,
+            channel,
-    ) -> Self {
+            fut_sender,
-        Self { client, tag_client }
+        }
+    }
+
+    fn create_output_client(&self) -> OutputServiceClient<Channel> {
+        OutputServiceClient::new(self.channel.clone())
+    }
+
+    fn create_tag_client(&self) -> TagServiceClient<Channel> {
+        TagServiceClient::new(self.channel.clone())
     }
 
     pub fn get_all(&self) -> impl Iterator<Item = OutputHandle> {
-        let mut client = self.client.clone();
+        let mut client = self.create_output_client();
-        let tag_client = self.tag_client.clone();
+        let tag_client = self.create_tag_client();
         block_on(client.get(output::v0alpha1::GetRequest {}))
             .unwrap()
             .into_inner()
@@ -52,37 +63,43 @@ impl Output {
             for_all(output);
         }
 
-        let mut client = self.client.clone();
+        let mut client = self.create_output_client();
-        let tag_client = self.tag_client.clone();
+        let tag_client = self.create_tag_client();
 
-        tokio::spawn(async move {
+        self.fut_sender
-            let mut stream = client
+            .unbounded_send(
-                .connect_for_all(ConnectForAllRequest {})
+                async move {
-                .await
+                    let mut stream = client
-                .unwrap()
+                        .connect_for_all(ConnectForAllRequest {})
-                .into_inner();
+                        .await
+                        .unwrap()
+                        .into_inner();
 
-            while let Some(Ok(response)) = stream.next().await {
+                    while let Some(Ok(response)) = stream.next().await {
-                let Some(output_name) = response.output_name else {
+                        let Some(output_name) = response.output_name else {
-                    continue;
+                            continue;
-                };
+                        };
 
-                let output = OutputHandle {
+                        let output = OutputHandle {
-                    client: client.clone(),
+                            client: client.clone(),
-                    tag_client: tag_client.clone(),
+                            tag_client: tag_client.clone(),
-                    name: output_name,
+                            name: output_name,
-                };
+                        };
 
-                for_all(output);
+                        for_all(output);
-            }
+                    }
-        });
+                }
+                .boxed(),
+            )
+            .unwrap();
     }
 }
 
+#[derive(Clone, Debug)]
 pub struct OutputHandle {
-    client: OutputServiceClient<Channel>,
+    pub(crate) client: OutputServiceClient<Channel>,
-    tag_client: TagServiceClient<Channel>,
+    pub(crate) tag_client: TagServiceClient<Channel>,
-    name: String,
+    pub(crate) name: String,
 }
 
 impl OutputHandle {
@@ -126,6 +143,7 @@ impl OutputHandle {
                 .into_iter()
                 .map(|id| TagHandle {
                     client: self.tag_client.clone(),
+                    output_client: self.client.clone(),
                     id,
                 })
                 .collect(),

api/rust_grpc/src/pinnacle.rs

@@ -6,16 +6,20 @@ use tonic::transport::Channel;
 
 #[derive(Debug, Clone)]
 pub struct Pinnacle {
-    client: PinnacleServiceClient<Channel>,
+    channel: Channel,
 }
 
 impl Pinnacle {
-    pub fn new(client: PinnacleServiceClient<Channel>) -> Self {
+    pub fn new(channel: Channel) -> Self {
-        Self { client }
+        Self { channel }
+    }
+
+    fn create_pinnacle_client(&self) -> PinnacleServiceClient<Channel> {
+        PinnacleServiceClient::new(self.channel.clone())
     }
 
     pub fn quit(&self) {
-        let mut client = self.client.clone();
+        let mut client = self.create_pinnacle_client();
         block_on(client.quit(QuitRequest {})).unwrap();
     }
 }

api/rust_grpc/src/process.rs

@@ -1,15 +1,13 @@
+use futures::{channel::mpsc::UnboundedSender, future::BoxFuture, FutureExt, StreamExt};
 use pinnacle_api_defs::pinnacle::process::v0alpha1::{
     process_service_client::ProcessServiceClient, SpawnRequest,
 };
-use tokio_stream::StreamExt;
 use tonic::transport::Channel;
 
-use crate::FutSender;
-
 #[derive(Debug, Clone)]
 pub struct Process {
-    client: ProcessServiceClient<Channel>,
+    channel: Channel,
-    fut_sender: FutSender,
+    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
 }
 
 pub struct SpawnCallbacks {
@@ -19,8 +17,15 @@ pub struct SpawnCallbacks {
 }
 
 impl Process {
-    pub fn new(client: ProcessServiceClient<Channel>, fut_sender: FutSender) -> Process {
+    pub fn new(channel: Channel, fut_sender: UnboundedSender<BoxFuture<'static, ()>>) -> Process {
-        Self { client, fut_sender }
+        Self {
+            channel,
+            fut_sender,
+        }
+    }
+
+    fn create_process_client(&self) -> ProcessServiceClient<Channel> {
+        ProcessServiceClient::new(self.channel.clone())
     }
 
     pub fn spawn(&self, args: impl IntoIterator<Item = impl Into<String>>) {
@@ -53,12 +58,9 @@ impl Process {
         once: bool,
         callbacks: Option<SpawnCallbacks>,
     ) {
-        let mut client = self.client.clone();
+        let mut client = self.create_process_client();
 
-        let args = args
+        let args = args.into_iter().map(Into::into).collect::<Vec<_>>();
-            .into_iter()
-            .map(|arg| Into::<String>::into(arg))
-            .collect::<Vec<_>>();
 
         let request = SpawnRequest {
             args,
@@ -67,49 +69,30 @@ impl Process {
         };
 
         self.fut_sender
-            .send(Box::pin(async move {
+            .unbounded_send(
-                let mut stream = client.spawn(request).await.unwrap().into_inner();
+                async move {
-                let Some(mut callbacks) = callbacks else { return };
+                    let mut stream = client.spawn(request).await.unwrap().into_inner();
-                while let Some(Ok(response)) = stream.next().await {
+                    let Some(mut callbacks) = callbacks else { return };
-                    if let Some(line) = response.stdout {
+                    while let Some(Ok(response)) = stream.next().await {
-                        if let Some(stdout) = callbacks.stdout.as_mut() {
+                        if let Some(line) = response.stdout {
-                            stdout(line);
+                            if let Some(stdout) = callbacks.stdout.as_mut() {
+                                stdout(line);
+                            }
                         }
-                    }
+                        if let Some(line) = response.stderr {
-                    if let Some(line) = response.stderr {
+                            if let Some(stderr) = callbacks.stderr.as_mut() {
-                        if let Some(stderr) = callbacks.stderr.as_mut() {
+                                stderr(line);
-                            stderr(line);
+                            }
                         }
-                    }
+                        if let Some(exit_msg) = response.exit_message {
-                    if let Some(exit_msg) = response.exit_message {
+                            if let Some(exit) = callbacks.exit.as_mut() {
-                        if let Some(exit) = callbacks.exit.as_mut() {
+                                exit(response.exit_code, exit_msg);
-                            exit(response.exit_code, exit_msg);
+                            }
                         }
                     }
                 }
-            }))
+                .boxed(),
+            )
             .unwrap();
-
-        // tokio::spawn(async move {
-        //     let mut stream = client.spawn(request).await.unwrap().into_inner();
-        //     let Some(mut callbacks) = callbacks else { return };
-        //     while let Some(Ok(response)) = stream.next().await {
-        //         if let Some(line) = response.stdout {
-        //             if let Some(stdout) = callbacks.stdout.as_mut() {
-        //                 stdout(line);
-        //             }
-        //         }
-        //         if let Some(line) = response.stderr {
-        //             if let Some(stderr) = callbacks.stderr.as_mut() {
-        //                 stderr(line);
-        //             }
-        //         }
-        //         if let Some(exit_msg) = response.exit_message {
-        //             if let Some(exit) = callbacks.exit.as_mut() {
-        //                 exit(response.exit_code, exit_msg);
-        //             }
-        //         }
-        //     }
-        // });
     }
 }

api/rust_grpc/src/tag.rs

@@ -1,27 +1,104 @@
-use futures::executor::block_on;
+use futures::{channel::mpsc::UnboundedSender, executor::block_on, future::BoxFuture};
 use num_enum::TryFromPrimitive;
-use pinnacle_api_defs::pinnacle::tag::{
+use pinnacle_api_defs::pinnacle::{
-    self,
+    output::v0alpha1::output_service_client::OutputServiceClient,
-    v0alpha1::{
+    tag::{
-        tag_service_client::TagServiceClient, RemoveRequest, SetActiveRequest, SetLayoutRequest,
+        self,
+        v0alpha1::{
+            tag_service_client::TagServiceClient, AddRequest, RemoveRequest, SetActiveRequest,
+            SetLayoutRequest, SwitchToRequest,
+        },
     },
 };
 use tonic::transport::Channel;
 
+use crate::output::{Output, OutputHandle};
+
 #[derive(Clone, Debug)]
 pub struct Tag {
-    client: TagServiceClient<Channel>,
+    channel: Channel,
+    fut_sender: UnboundedSender<BoxFuture<'static, ()>>,
 }
 
 impl Tag {
-    pub fn new(client: TagServiceClient<Channel>) -> Self {
+    pub fn new(channel: Channel, fut_sender: UnboundedSender<BoxFuture<'static, ()>>) -> Self {
-        Self { client }
+        Self {
+            channel,
+            fut_sender,
+        }
+    }
+
+    pub fn create_tag_client(&self) -> TagServiceClient<Channel> {
+        TagServiceClient::new(self.channel.clone())
+    }
+
+    pub fn create_output_client(&self) -> OutputServiceClient<Channel> {
+        OutputServiceClient::new(self.channel.clone())
+    }
+
+    pub fn add(
+        &self,
+        output: &OutputHandle,
+        tag_names: impl IntoIterator<Item = impl Into<String>>,
+    ) -> impl Iterator<Item = TagHandle> {
+        let mut client = self.create_tag_client();
+        let output_client = self.create_output_client();
+
+        let tag_names = tag_names.into_iter().map(Into::into).collect();
+
+        let response = block_on(client.add(AddRequest {
+            output_name: Some(output.name.clone()),
+            tag_names,
+        }))
+        .unwrap()
+        .into_inner();
+
+        response.tag_ids.into_iter().map(move |id| TagHandle {
+            client: client.clone(),
+            output_client: output_client.clone(),
+            id,
+        })
+    }
+
+    pub fn get_all(&self) -> impl Iterator<Item = TagHandle> {
+        let mut client = self.create_tag_client();
+        let output_client = self.create_output_client();
+
+        let response = block_on(client.get(tag::v0alpha1::GetRequest {}))
+            .unwrap()
+            .into_inner();
+
+        response.tag_ids.into_iter().map(move |id| TagHandle {
+            client: client.clone(),
+            output_client: output_client.clone(),
+            id,
+        })
+    }
+
+    pub fn get(&self, name: impl Into<String>, output: Option<&OutputHandle>) -> Option<TagHandle> {
+        let name = name.into();
+        let output_module = Output::new(self.channel.clone(), self.fut_sender.clone());
+
+        self.get_all().find(|tag| {
+            let props = tag.props();
+
+            let same_tag_name = props.name.as_ref() == Some(&name);
+            let same_output = props.output.is_some_and(|op| {
+                Some(op.name)
+                    == output
+                        .map(|o| o.name.clone())
+                        .or_else(|| output_module.get_focused().map(|o| o.name))
+            });
+
+            same_tag_name && same_output
+        })
     }
 }
 
 #[derive(Debug, Clone)]
 pub struct TagHandle {
     pub(crate) client: TagServiceClient<Channel>,
+    pub(crate) output_client: OutputServiceClient<Channel>,
     pub(crate) id: u32,
 }
 
@@ -72,13 +149,38 @@ impl TagHandle {
         .unwrap();
     }
 
+    pub fn switch_to(&self) {
+        let mut client = self.client.clone();
+        block_on(client.switch_to(SwitchToRequest {
+            tag_id: Some(self.id),
+        }))
+        .unwrap();
+    }
+
     pub fn props(&self) -> TagProperties {
-        todo!()
+        let mut client = self.client.clone();
+        let output_client = self.output_client.clone();
+
+        let response = block_on(client.get_properties(tag::v0alpha1::GetPropertiesRequest {
+            tag_id: Some(self.id),
+        }))
+        .unwrap()
+        .into_inner();
+
+        TagProperties {
+            active: response.active,
+            name: response.name,
+            output: response.output_name.map(|name| OutputHandle {
+                client: output_client,
+                tag_client: client,
+                name,
+            }),
+        }
     }
 }
 
 pub struct TagProperties {
     pub active: Option<bool>,
     pub name: Option<String>,
-    pub output: (),
+    pub output: Option<OutputHandle>,
 }

api/rust_grpc/src/window.rs

@@ -1,9 +1,10 @@
 use futures::executor::block_on;
 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, MoveToTagRequest, SetTagRequest,
+        window_service_client::WindowServiceClient, CloseRequest, MoveToTagRequest, SetTagRequest,
     },
     window::{
         self,
@@ -19,22 +20,47 @@ use crate::{input::MouseButton, tag::TagHandle, util::Geometry};
 
 #[derive(Debug, Clone)]
 pub struct Window {
-    client: WindowServiceClient<Channel>,
+    channel: Channel,
-    tag_client: TagServiceClient<Channel>,
 }
 
 impl Window {
-    pub fn new(
+    pub fn new(channel: Channel) -> Self {
-        client: WindowServiceClient<Channel>,
+        Self { channel }
-        tag_client: TagServiceClient<Channel>,
+    }
-    ) -> Self {
+
-        Self { client, tag_client }
+    pub fn create_window_client(&self) -> WindowServiceClient<Channel> {
+        WindowServiceClient::new(self.channel.clone())
+    }
+
+    pub fn create_tag_client(&self) -> TagServiceClient<Channel> {
+        TagServiceClient::new(self.channel.clone())
+    }
+
+    pub fn create_output_client(&self) -> OutputServiceClient<Channel> {
+        OutputServiceClient::new(self.channel.clone())
+    }
+
+    pub fn begin_move(&self, button: MouseButton) {
+        let mut client = self.create_window_client();
+        block_on(client.move_grab(MoveGrabRequest {
+            button: Some(button as u32),
+        }))
+        .unwrap();
+    }
+
+    pub fn begin_resize(&self, button: MouseButton) {
+        let mut client = self.create_window_client();
+        block_on(client.resize_grab(ResizeGrabRequest {
+            button: Some(button as u32),
+        }))
+        .unwrap();
     }
 
     /// Get all windows.
     pub fn get_all(&self) -> impl Iterator<Item = WindowHandle> {
-        let mut client = self.client.clone();
+        let mut client = self.create_window_client();
-        let tag_client = self.tag_client.clone();
+        let tag_client = self.create_tag_client();
+        let output_client = self.create_output_client();
         block_on(client.get(GetRequest {}))
             .unwrap()
             .into_inner()
@@ -44,6 +70,7 @@ impl Window {
                 client: client.clone(),
                 id,
                 tag_client: tag_client.clone(),
+                output_client: output_client.clone(),
             })
     }
 
@@ -59,6 +86,7 @@ pub struct WindowHandle {
     pub(crate) client: WindowServiceClient<Channel>,
     pub(crate) id: u32,
     pub(crate) tag_client: TagServiceClient<Channel>,
+    pub(crate) output_client: OutputServiceClient<Channel>,
 }
 
 #[repr(i32)]
@@ -81,7 +109,14 @@ pub struct WindowProperties {
 }
 
 impl WindowHandle {
-    pub fn set_fullscreen(&mut self, set: bool) {
+    pub fn close(mut self) {
+        block_on(self.client.close(CloseRequest {
+            window_id: Some(self.id),
+        }))
+        .unwrap();
+    }
+
+    pub fn set_fullscreen(&self, set: bool) {
         let mut client = self.client.clone();
         block_on(client.set_fullscreen(SetFullscreenRequest {
             window_id: Some(self.id),
@@ -92,7 +127,7 @@ impl WindowHandle {
         .unwrap();
     }
 
-    pub fn toggle_fullscreen(&mut self) {
+    pub fn toggle_fullscreen(&self) {
         let mut client = self.client.clone();
         block_on(client.set_fullscreen(SetFullscreenRequest {
             window_id: Some(self.id),
@@ -101,7 +136,7 @@ impl WindowHandle {
         .unwrap();
     }
 
-    pub fn set_maximized(&mut self, set: bool) {
+    pub fn set_maximized(&self, set: bool) {
         let mut client = self.client.clone();
         block_on(client.set_maximized(SetMaximizedRequest {
             window_id: Some(self.id),
@@ -112,7 +147,7 @@ impl WindowHandle {
         .unwrap();
     }
 
-    pub fn toggle_maximized(&mut self) {
+    pub fn toggle_maximized(&self) {
         let mut client = self.client.clone();
         block_on(client.set_maximized(SetMaximizedRequest {
             window_id: Some(self.id),
@@ -121,7 +156,7 @@ impl WindowHandle {
         .unwrap();
     }
 
-    pub fn set_floating(&mut self, set: bool) {
+    pub fn set_floating(&self, set: bool) {
         let mut client = self.client.clone();
         block_on(client.set_floating(SetFloatingRequest {
             window_id: Some(self.id),
@@ -132,7 +167,7 @@ impl WindowHandle {
         .unwrap();
     }
 
-    pub fn toggle_floating(&mut self) {
+    pub fn toggle_floating(&self) {
         let mut client = self.client.clone();
         block_on(client.set_floating(SetFloatingRequest {
             window_id: Some(self.id),
@@ -175,22 +210,6 @@ impl WindowHandle {
         .unwrap();
     }
 
-    pub fn begin_move(&self, button: MouseButton) {
-        let mut client = self.client.clone();
-        block_on(client.move_grab(MoveGrabRequest {
-            button: Some(button as u32),
-        }))
-        .unwrap();
-    }
-
-    pub fn begin_resize(&self, button: MouseButton) {
-        let mut client = self.client.clone();
-        block_on(client.resize_grab(ResizeGrabRequest {
-            button: Some(button as u32),
-        }))
-        .unwrap();
-    }
-
     pub fn props(&self) -> WindowProperties {
         let mut client = self.client.clone();
         let tag_client = self.tag_client.clone();
@@ -227,6 +246,7 @@ impl WindowHandle {
                 .into_iter()
                 .map(|id| TagHandle {
                     client: tag_client.clone(),
+                    output_client: self.output_client.clone(),
                     id,
                 })
                 .collect(),