Merged crates, more work on api

Cargo.toml

@@ -1,5 +1,37 @@
-[workspace]
+[package]
-members = [
+name = "pinnacle"
-    "pinnacle",
+version = "0.1.0"
-    "pinnacle_api",
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+tracing = "0.1.37"
+tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
+smithay = { git = "https://github.com/Smithay/smithay", rev = "9b3d173" }
+smithay-drm-extras = { git = "https://github.com/Smithay/smithay", optional = true }
+thiserror = "1.0.40"
+xcursor = {version = "0.3.4", optional = true }
+image = {version = "0.24.0", default-features = false, optional = true}
+serde = { version = "1.0.164", features = ["derive"] }
+rmp = { version = "0.8.11" }
+rmp-serde = { version = "1.1.1" }
+
+[features]
+default = ["egl", "winit", "udev"]
+egl = ["smithay/use_system_lib", "smithay/backend_egl"]
+udev = [
+  "smithay-drm-extras",
+  "smithay/backend_libinput",
+  "smithay/backend_udev",
+  "smithay/backend_drm",
+  "smithay/backend_gbm",
+  "smithay/backend_vulkan",
+  "smithay/backend_egl",
+  "smithay/backend_session_libseat",
+  "image",
+  "smithay/renderer_gl",
+  "smithay/renderer_multi",
+  "xcursor",
 ]
+winit = ["smithay/backend_winit", "smithay/backend_drm"]

pinnacle/Cargo.toml

@@ -1,35 +0,0 @@
-[package]
-name = "pinnacle"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-pinnacle_api = { path = "../pinnacle_api" }
-tracing = "0.1.37"
-tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
-smithay = { git = "https://github.com/Smithay/smithay", rev = "9b3d173" }
-smithay-drm-extras = { git = "https://github.com/Smithay/smithay", optional = true }
-thiserror = "1.0.40"
-xcursor = {version = "0.3.4", optional = true }
-image = {version = "0.24.0", default-features = false, optional = true}
-
-[features]
-default = ["egl", "winit", "udev"]
-egl = ["smithay/use_system_lib", "smithay/backend_egl"]
-udev = [
-  "smithay-drm-extras",
-  "smithay/backend_libinput",
-  "smithay/backend_udev",
-  "smithay/backend_drm",
-  "smithay/backend_gbm",
-  "smithay/backend_vulkan",
-  "smithay/backend_egl",
-  "smithay/backend_session_libseat",
-  "image",
-  "smithay/renderer_gl",
-  "smithay/renderer_multi",
-  "xcursor",
-]
-winit = ["smithay/backend_winit", "smithay/backend_drm"]

pinnacle_api/Cargo.toml

@@ -1,15 +0,0 @@
-[package]
-name = "pinnacle_api"
-version = "0.1.0"
-edition = "2021"
-
-[dependencies]
-tracing = "0.1.37"
-tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
-rkyv = { version = "0.7.42", features = ["validation", "strict"] }
-num_enum = "0.6.1"
-serde = { version = "1.0.164", features = ["derive"] }
-serde_json = { version = "1.0.96" }
-rmp-serde = { version = "1.1.1" }
-xkbcommon = { version = "0.5.0" }
-smithay = { git = "https://github.com/Smithay/smithay", rev = "9b3d173" }

pinnacle_api/src/lib.rs

@@ -1,189 +0,0 @@
-use std::{
-    error::Error,
-    io,
-    os::unix::net::{UnixListener, UnixStream},
-    path::Path,
-};
-
-use smithay::reexports::calloop::{
-    self, generic::Generic, EventSource, Interest, Mode, PostAction,
-};
-
-use crate::message::Msg;
-
-pub mod message;
-
-const SOCKET_PATH: &str = "/tmp/pinnacle_socket";
-
-pub fn run() -> Result<(), Box<dyn Error>> {
-    let socket_path = Path::new(SOCKET_PATH);
-
-    if socket_path.exists() {
-        std::fs::remove_file(socket_path)?;
-    }
-
-    let listener = UnixListener::bind(SOCKET_PATH)?;
-
-    std::thread::spawn(move || {
-        for stream in listener.incoming() {
-            match stream {
-                Ok(stream) => {
-                    std::thread::spawn(|| handle_client(stream));
-                }
-                Err(err) => {
-                    eprintln!("Incoming stream error: {}", err);
-                }
-            }
-        }
-    });
-
-    Ok(())
-}
-
-fn handle_client(stream: UnixStream) {
-    loop {
-        let msg: Msg = rmp_serde::from_read(&stream).unwrap();
-
-        println!("{:?}", msg);
-    }
-}
-
-pub struct PinnacleSocketSource {
-    socket: Generic<UnixListener>,
-}
-
-impl PinnacleSocketSource {
-    pub fn new() -> Result<Self, io::Error> {
-        let socket_path = Path::new(SOCKET_PATH);
-
-        if socket_path.exists() {
-            std::fs::remove_file(socket_path)?;
-        }
-
-        let listener = UnixListener::bind(SOCKET_PATH)?;
-
-        let socket = Generic::new(listener, Interest::READ, Mode::Level);
-
-        Ok(Self { socket })
-    }
-}
-
-impl EventSource for PinnacleSocketSource {
-    type Event = UnixStream;
-
-    type Metadata = ();
-
-    type Ret = ();
-
-    type Error = io::Error;
-
-    fn process_events<F>(
-        &mut self,
-        readiness: calloop::Readiness,
-        token: calloop::Token,
-        mut callback: F,
-    ) -> Result<calloop::PostAction, Self::Error>
-    where
-        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
-    {
-        self.socket
-            .process_events(readiness, token, |_readiness, listener| {
-                listener.set_nonblocking(true)?;
-                while let Ok((stream, _sock_addr)) = listener.accept() {
-                    stream.set_nonblocking(true)?;
-                    callback(stream, &mut ());
-                }
-
-                Ok(PostAction::Continue)
-            })
-    }
-
-    fn register(
-        &mut self,
-        poll: &mut calloop::Poll,
-        token_factory: &mut calloop::TokenFactory,
-    ) -> calloop::Result<()> {
-        self.socket.register(poll, token_factory)
-    }
-
-    fn reregister(
-        &mut self,
-        poll: &mut calloop::Poll,
-        token_factory: &mut calloop::TokenFactory,
-    ) -> calloop::Result<()> {
-        self.socket.reregister(poll, token_factory)
-    }
-
-    fn unregister(&mut self, poll: &mut calloop::Poll) -> calloop::Result<()> {
-        self.socket.unregister(poll)
-    }
-}
-
-pub struct PinnacleStreamSource {
-    stream: Generic<UnixStream>,
-}
-
-impl PinnacleStreamSource {
-    pub fn new(stream: UnixStream) -> Self {
-        Self {
-            stream: Generic::new(stream, Interest::READ, Mode::Level),
-        }
-    }
-}
-
-impl EventSource for PinnacleStreamSource {
-    type Event = Msg;
-
-    type Metadata = ();
-
-    type Ret = ();
-
-    type Error = io::Error;
-
-    fn process_events<F>(
-        &mut self,
-        readiness: calloop::Readiness,
-        token: calloop::Token,
-        mut callback: F,
-    ) -> Result<PostAction, Self::Error>
-    where
-        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
-    {
-        self.stream
-            .process_events(readiness, token, |_readiness, stream| {
-                match rmp_serde::from_read(stream as &UnixStream) {
-                    Ok(msg) => callback(msg, &mut ()),
-                    Err(rmp_serde::decode::Error::InvalidMarkerRead(err))
-                        if err.kind() == io::ErrorKind::UnexpectedEof =>
-                    {
-                        stream.shutdown(std::net::Shutdown::Both)?;
-                        println!("Stream closed: {:?}", err);
-                        return Ok(PostAction::Remove);
-                    }
-                    Err(err) => println!("{:?}", err),
-                }
-
-                Ok(PostAction::Continue)
-            })
-    }
-
-    fn register(
-        &mut self,
-        poll: &mut calloop::Poll,
-        token_factory: &mut calloop::TokenFactory,
-    ) -> calloop::Result<()> {
-        self.stream.register(poll, token_factory)
-    }
-
-    fn reregister(
-        &mut self,
-        poll: &mut calloop::Poll,
-        token_factory: &mut calloop::TokenFactory,
-    ) -> calloop::Result<()> {
-        self.stream.reregister(poll, token_factory)
-    }
-
-    fn unregister(&mut self, poll: &mut calloop::Poll) -> calloop::Result<()> {
-        self.stream.unregister(poll)
-    }
-}

pinnacle_api_lua/README.md

@@ -1,2 +1,2 @@
-The cffi library courtesy of https://github.com/q66/cffi-lua.
 The luaposix library courtesy of https://github.com/luaposix/luaposix.
+NOTE: there is an issue with this library where recv's return value is size_t, not ssize_t as it should be. This library uses a patched version, see https://github.com/luaposix/luaposix/issues/354.

pinnacle_api_lua/init.lua

@@ -1,29 +1,35 @@
 -- require("luarocks.loader")
 
-package.path = "./lib/?.lua;./lib/?/init.lua;" .. package.path
+local LOCAL_PATH = "/home/jason/projects/pinnacle/pinnacle_api_lua"
-package.cpath = "./lib/?.so;" .. package.cpath
+
+package.path = LOCAL_PATH .. "/lib/?.lua;" .. LOCAL_PATH .. "/lib/?/init.lua;" .. package.path
+package.cpath = LOCAL_PATH .. "/lib/?.so;" .. package.cpath
 
 local socket = require("posix.sys.socket")
-local msgpack = require("lib.msgpack")
+local fcntl = require("posix.fcntl")
+local msgpack = require("msgpack")
 
 local SOCKET_PATH = "/tmp/pinnacle_socket"
 
-local CONFIG_PATH = os.getenv("XDG_CONFIG_HOME") .. "/pinnacle/init.lua"
+local CONFIG_PATH = (os.getenv("XDG_CONFIG_HOME") or "~/.config") .. "/pinnacle/init.lua"
 
 package.path = CONFIG_PATH .. ";" .. package.path
 
-local sockaddr = {
+---@type integer
-    family = socket.AF_UNIX,
-    path = SOCKET_PATH,
-}
-
 local socket_fd = assert(socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0), "Failed to create socket")
 print("created socket at fd " .. socket_fd)
 
-assert(0 == socket.connect(socket_fd, sockaddr), "Failed to connect to Pinnacle socket")
+assert(0 == socket.connect(socket_fd, {
+    family = socket.AF_UNIX,
+    path = SOCKET_PATH,
+}), "Failed to connect to Pinnacle socket")
 
 function SendMsg(data)
-    socket.send(socket_fd, msgpack.encode(data))
+    local encoded = msgpack.encode(data)
+    assert(encoded)
+    local len = encoded:len()
+    socket.send(socket_fd, string.pack("=I4", len))
+    socket.send(socket_fd, encoded)
 end
 
 ---@type function[]
@@ -31,24 +37,51 @@ CallbackTable = {}
 
 assert(pcall(require, "pinnacle"), "config file not found")
 
--- local str = msgpack.encode({
+---Read the specified number of bytes.
---     SetMousebind = { button = 6 },
+---@param socket_fd integer The socket file descriptor
--- })
+---@param count integer The amount of bytes to read
--- local str = msgpack.encode({
+---@return string|nil data
---     SetKeybind = {
+---@return string|nil err_msg
---         key = "This is a key",
+---@return integer|nil err_num
---         modifiers = { "ctrl", "boogers", "numpty" },
+local function read_exact(socket_fd, count)
---     },
+    local len_to_read = count
--- })
+    local data = ""
--- print(str)
+    while len_to_read > 0 do
---
+        local bytes, err_msg, errnum = socket.recv(socket_fd, len_to_read)
--- socket.send(socket_fd, str)
 
--- unistd.close(socket_fd)
+        if bytes == nil then
+            -- TODO: handle errors
+            print("bytes was nil")
+            return bytes, err_msg, errnum
+        end
 
--- local keys = require("keys")
+        ---@type integer
---
+        local recv_len = bytes:len()
--- local input = require("input")
+
--- input.keybind({ "Shift", "Ctrl" }, keys.c, "CloseWindow")
+        if recv_len == 0 then
-while true do
+            print("stream closed")
+            break
+        end
+
+        len_to_read = len_to_read - recv_len
+        assert(len_to_read >= 0, "Overread message boundary")
+
+        data = data .. bytes
+    end
+    return data
+end
+
+-- TODO: set timeouts so that you actually make sure the msg is correct
+while true do
+    local msg_len_bytes, err_msg, err_num = read_exact(socket_fd, 4)
+    assert(msg_len_bytes)
+
+    ---@type integer
+    local msg_len = string.unpack("=I4", msg_len_bytes)
+
+    local msg_bytes, err_msg2, err_num2 = read_exact(socket_fd, msg_len)
+    assert(msg_bytes)
+
+    local tb = msgpack.decode(msg_bytes)
+    print(tb)
 end

pinnacle_api_lua/lib/msgpack.lua

@@ -346,7 +346,12 @@ return {
         end
     end,
 
-    -- primary decode function
+    -- Decode a string of MessagePack data into objects.
+    --
+    ---@param data string The MessagePack packet
+    ---@param position any
+    ---@return any ... The decoded values unpacked. If decoding failed, nil
+    ---@return string | nil error The string "cannot decode MessagePack" if decoding failed
     decode = function(data, position)
         local values, value, ok = {}
         position = position or 1

src/api.rs

@@ -0,0 +1,144 @@
+pub mod msg;
+
+use std::{
+    io::{self, Read, Write},
+    os::unix::net::{UnixDatagram, UnixListener, UnixStream},
+    path::Path,
+};
+
+use smithay::reexports::calloop::{
+    self, channel::Sender, generic::Generic, EventSource, Interest, Mode, PostAction,
+};
+
+use self::msg::{Msg, OutgoingMsg};
+
+const SOCKET_PATH: &str = "/tmp/pinnacle_socket";
+
+fn handle_client(mut stream: UnixStream, sender: Sender<Msg>) {
+    loop {
+        let mut len_marker_bytes = [0u8; 4];
+        if let Err(err) = stream.read_exact(&mut len_marker_bytes) {
+            if err.kind() == io::ErrorKind::UnexpectedEof {
+                tracing::warn!("stream closed: {}", err);
+                stream.shutdown(std::net::Shutdown::Both).unwrap();
+                break;
+            }
+        };
+
+        let len_marker = u32::from_ne_bytes(len_marker_bytes);
+        let mut msg_bytes = vec![0u8; len_marker as usize];
+
+        if let Err(err) = stream.read_exact(msg_bytes.as_mut_slice()) {
+            if err.kind() == io::ErrorKind::UnexpectedEof {
+                tracing::warn!("stream closed: {}", err);
+                stream.shutdown(std::net::Shutdown::Both).unwrap();
+                break;
+            }
+        };
+        let msg: Msg = rmp_serde::from_slice(msg_bytes.as_slice()).unwrap(); // TODO: handle error
+
+        sender.send(msg).unwrap();
+    }
+    tracing::info!("end of handle_client");
+}
+
+pub struct PinnacleSocketSource {
+    socket: Generic<UnixListener>,
+    sender: Sender<Msg>,
+}
+
+impl PinnacleSocketSource {
+    pub fn new(sender: Sender<Msg>) -> Result<Self, io::Error> {
+        let socket_path = Path::new(SOCKET_PATH);
+
+        if let Ok(exists) = socket_path.try_exists() {
+            if exists {
+                std::fs::remove_file(socket_path)?;
+            }
+        }
+
+        let listener = UnixListener::bind(SOCKET_PATH)?;
+        listener.set_nonblocking(true)?;
+
+        let socket = Generic::new(listener, Interest::READ, Mode::Level);
+
+        Ok(Self { socket, sender })
+    }
+}
+
+pub fn send_to_client(
+    stream: &mut UnixStream,
+    msg: &OutgoingMsg,
+) -> Result<(), rmp_serde::encode::Error> {
+    let msg = rmp_serde::to_vec_named(msg)?;
+    let msg_len = msg.len() as u32;
+    let bytes = msg_len.to_ne_bytes();
+
+    if let Err(err) = stream.write_all(&bytes) {
+        if err.kind() == io::ErrorKind::BrokenPipe {
+            // TODO: notify user that config daemon is ded
+            return Ok(()); // TODO:
+        }
+    }
+    if let Err(err) = stream.write_all(msg.as_slice()) {
+        if err.kind() == io::ErrorKind::BrokenPipe {
+            // TODO: something
+            return Ok(()); // TODO:
+        }
+    };
+    Ok(())
+}
+
+impl EventSource for PinnacleSocketSource {
+    type Event = UnixStream;
+
+    type Metadata = ();
+
+    type Ret = ();
+
+    type Error = io::Error;
+
+    fn process_events<F>(
+        &mut self,
+        readiness: calloop::Readiness,
+        token: calloop::Token,
+        mut callback: F,
+    ) -> Result<calloop::PostAction, Self::Error>
+    where
+        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
+    {
+        self.socket
+            .process_events(readiness, token, |_readiness, listener| {
+                while let Ok((stream, _sock_addr)) = listener.accept() {
+                    let sender = self.sender.clone();
+                    let callback_stream = stream.try_clone().unwrap(); // TODO: error
+                    callback(callback_stream, &mut ());
+                    std::thread::spawn(move || {
+                        handle_client(stream, sender);
+                    });
+                }
+
+                Ok(PostAction::Continue)
+            })
+    }
+
+    fn register(
+        &mut self,
+        poll: &mut calloop::Poll,
+        token_factory: &mut calloop::TokenFactory,
+    ) -> calloop::Result<()> {
+        self.socket.register(poll, token_factory)
+    }
+
+    fn reregister(
+        &mut self,
+        poll: &mut calloop::Poll,
+        token_factory: &mut calloop::TokenFactory,
+    ) -> calloop::Result<()> {
+        self.socket.reregister(poll, token_factory)
+    }
+
+    fn unregister(&mut self, poll: &mut calloop::Poll) -> calloop::Result<()> {
+        self.socket.unregister(poll)
+    }
+}

src/api/msg.rs

@@ -41,3 +41,9 @@ impl<T: IntoIterator<Item = Modifiers>> From<T> for ModifierMask {
         Self(mask)
     }
 }
+
+/// Messages sent from the server to each client.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub enum OutgoingMsg {
+    CallCallback(u32),
+}

src/backend/winit.rs

@@ -138,7 +138,7 @@ pub fn run_winit() -> Result<(), Box<dyn Error>> {
             Some(dmabuf_default_feedback)
         }
         Ok(None) => {
-            tracing::warn!("failed to query render node, dmabuf will use v3"); // TODO: tracing
+            tracing::warn!("failed to query render node, dmabuf will use v3");
             None
         }
         Err(err) => {
@@ -181,6 +181,11 @@ pub fn run_winit() -> Result<(), Box<dyn Error>> {
         evt_loop_handle,
     )?;
 
+    // std::process::Command::new("lua")
+    //     .arg("../pinnacle_api_lua/init.lua")
+    //     .spawn()
+    //     .unwrap();
+
     state
         .shm_state
         .update_formats(state.backend_data.backend.renderer().shm_formats());

src/input.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use pinnacle_api::message::{ModifierMask, Modifiers};
+use crate::api::msg::{ModifierMask, Modifiers, OutgoingMsg};
 use smithay::{
     backend::input::{
         AbsolutePositionEvent, Axis, AxisSource, ButtonState, Event, InputBackend, InputEvent,
@@ -216,13 +216,13 @@ impl<B: Backend> State<B> {
                         .keybinds
                         .get(&(modifier_mask.into(), keysym.modified_sym()))
                     {
-                        return FilterResult::Intercept(1);
+                        return FilterResult::Intercept(*callback_id);
                     }
                     match keysym.modified_sym() {
-                        keysyms::KEY_L => return FilterResult::Intercept(1),
+                        keysyms::KEY_L => return FilterResult::Intercept(100),
-                        keysyms::KEY_K => return FilterResult::Intercept(2),
+                        keysyms::KEY_K => return FilterResult::Intercept(200),
-                        keysyms::KEY_J => return FilterResult::Intercept(3),
+                        keysyms::KEY_J => return FilterResult::Intercept(300),
-                        keysyms::KEY_H => return FilterResult::Intercept(4),
+                        keysyms::KEY_H => return FilterResult::Intercept(400),
                         keysyms::KEY_Escape => {
                             state.loop_signal.stop();
                             return FilterResult::Intercept(0);
@@ -250,11 +250,22 @@ impl<B: Backend> State<B> {
         self.move_mode = move_mode;
 
         let program = match action {
-            Some(1) => "alacritty",
+            Some(100) => "alacritty",
-            Some(2) => "nautilus",
+            Some(200) => "nautilus",
-            Some(3) => "kitty",
+            Some(300) => "kitty",
-            Some(4) => "foot",
+            Some(400) => "foot",
-            Some(_) | None => return,
+            Some(callback_id) => {
+                if let Some(stream) = self.api_state.stream.as_mut() {
+                    if let Err(err) =
+                        crate::api::send_to_client(stream, &OutgoingMsg::CallCallback(callback_id))
+                    {
+                        // TODO: print error
+                    }
+                }
+
+                return;
+            }
+            None => return,
         };
 
         tracing::info!("Spawning {}", program);

src/main.rs

@@ -1,3 +1,4 @@
+mod api;
 mod backend;
 mod cursor;
 mod grab;

src/state.rs

@@ -1,6 +1,11 @@
-use std::{collections::HashMap, error::Error, os::fd::AsRawFd, sync::Arc};
+use std::{
+    collections::HashMap,
+    error::Error,
+    os::{fd::AsRawFd, unix::net::UnixStream},
+    sync::Arc,
+};
 
-use pinnacle_api::{message::Msg, PinnacleSocketSource, PinnacleStreamSource};
+use crate::api::{msg::Msg, PinnacleSocketSource, PinnacleStreamSource};
 use smithay::{
     backend::renderer::element::RenderElementStates,
     desktop::{
@@ -13,7 +18,10 @@ use smithay::{
     input::{keyboard::XkbConfig, pointer::CursorImageStatus, Seat, SeatState},
     output::Output,
     reexports::{
-        calloop::{generic::Generic, Interest, LoopHandle, LoopSignal, Mode, PostAction},
+        calloop::{
+            self, channel::Event, generic::Generic, Interest, LoopHandle, LoopSignal, Mode,
+            PostAction,
+        },
         wayland_server::{
             backend::{ClientData, ClientId, DisconnectReason},
             protocol::wl_surface::WlSurface,
@@ -59,6 +67,7 @@ pub struct State<B: Backend> {
     pub viewporter_state: ViewporterState,
     pub fractional_scale_manager_state: FractionalScaleManagerState,
     pub input_state: InputState,
+    pub api_state: ApiState,
 
     pub popup_manager: PopupManager,
 
@@ -97,29 +106,58 @@ impl<B: Backend> State<B> {
             },
         )?;
 
-        loop_handle.insert_source(PinnacleSocketSource::new()?, |stream, _, data| {
+        let (tx_channel, rx_channel) = calloop::channel::channel::<Msg>();
-            data.state
+        loop_handle.insert_source(rx_channel, |msg, _, data| match msg {
-                .loop_handle
+            Event::Msg(msg) => {
-                .insert_source(PinnacleStreamSource::new(stream), |msg, _, data| {
+                match msg {
-                    // TODO: do stuff with msg
+                    Msg::SetKeybind {
-                    match msg {
+                        key,
-                        Msg::SetKeybind {
+                        modifiers,
-                            key,
+                        callback_id,
-                            modifiers,
+                    } => {
-                            callback_id,
+                        tracing::info!("set keybind: {:?}, {}", modifiers, key);
-                        } => {
+                        data.state
-                            tracing::info!("set keybind: {:?}, {}", modifiers, key);
+                            .input_state
-                            data.state
+                            .keybinds
-                                .input_state
+                            .insert((modifiers.into(), key), callback_id);
-                                .keybinds
+                    }
-                                .insert((modifiers.into(), key), callback_id);
+                    Msg::SetMousebind { button } => todo!(),
-                        }
+                };
-                        Msg::SetMousebind { button } => todo!(),
+            }
-                    };
+            Event::Closed => todo!(),
-                })
-                .unwrap();
         })?;
 
+        // std::thread::spawn(move || {
+        //     crate::api::init_api_socket(tx_channel).unwrap();
+        // });
+
+        loop_handle.insert_source(PinnacleSocketSource::new(tx_channel)?, |stream, _, data| {
+            if let Some(old_stream) = data.state.api_state.stream.replace(stream) {
+                old_stream.shutdown(std::net::Shutdown::Both).unwrap();
+            }
+        })?;
+        //     data.state
+        //         .loop_handle
+        //         .insert_source(PinnacleStreamSource::new(stream), |msg, _, data| {
+        //             // TODO: do stuff with msg
+        //             match msg {
+        //                 Msg::SetKeybind {
+        //                     key,
+        //                     modifiers,
+        //                     callback_id,
+        //                 } => {
+        //                     tracing::info!("set keybind: {:?}, {}", modifiers, key);
+        //                     data.state
+        //                         .input_state
+        //                         .keybinds
+        //                         .insert((modifiers.into(), key), callback_id);
+        //                 }
+        //                 Msg::SetMousebind { button } => todo!(),
+        //             };
+        //         })
+        //         .unwrap();
+        // })?;
+
         let display_handle = display.handle();
         let mut seat_state = SeatState::new();
         let mut seat = seat_state.new_wl_seat(&display_handle, backend_data.seat_name());
@@ -147,6 +185,7 @@ impl<B: Backend> State<B> {
             input_state: InputState {
                 keybinds: HashMap::new(),
             },
+            api_state: ApiState { stream: None },
 
             seat,
 
@@ -165,6 +204,7 @@ impl<B: Backend> State<B> {
             .cloned()
     }
 
+    // TODO:
     pub fn handle_msg(msg: Msg) {
         match msg {
             Msg::SetKeybind {
@@ -231,3 +271,7 @@ pub fn take_presentation_feedback(
 
     output_presentation_feedback
 }
+
+pub struct ApiState {
+    pub stream: Option<UnixStream>,
+}