xpc: hacky implementation of GetVersion

kpcli/Cargo.toml

@@ -32,5 +32,7 @@ dbus-codegen = "0.10.0"
 
 # XPC (libxpc) interface only on macOS
 [target.'cfg(target_os = "macos")'.dependencies]
-futures-preview = "=0.2.2"
+block = "0.1.6"
+futures = "0.3.4"
 xpc-connection  = { git = "https://github.com/dfrankland/xpc-connection-rs.git", rev = "cd4fb3d", package = "xpc-connection" }
+xpc-connection-sys = { git = "https://github.com/dfrankland/xpc-connection-rs.git", rev = "cd4fb3d", package = "xpc-connection-sys" }

kpcli/src/daemon/xpc.rs

@@ -1,16 +1,107 @@
-#![cfg(target_os = "macos")]
-//! macOS XPC implementation of the DaemonInterface for kpcli.
-
 use super::{ConfigCommands, DaemonInterface};
 use anyhow::Result;
 use async_trait::async_trait;
-use futures::stream::StreamExt;
-use futures::executor::block_on;
+use futures_util::StreamExt;
 use std::collections::HashMap;
-use xpc_connection::{Message, XpcConnection};
+use std::ffi::{CStr, CString};
+use std::ops::Deref;
+use std::{pin::Pin, task::Poll};
+
+use xpc_connection::Message;
+
+use futures::{
+    channel::mpsc::{unbounded as unbounded_channel, UnboundedReceiver, UnboundedSender},
+    Stream,
+};
+
 
 const SERVICE_NAME: &str = "net.buzzert.kordophonecd\0";
-const GET_VERSION_METHOD: &str = "GetVersion\0";
+const GET_VERSION_METHOD: &str = "GetVersion";
+
+// We can't use XPCClient from xpc-connection because of some strange decisions with which flags
+// are passed to xpc_connection_create_mach_service.
+struct XPCClient {
+    connection: xpc_connection_sys::xpc_connection_t,
+    receiver: UnboundedReceiver<Message>,
+    sender: UnboundedSender<Message>,
+    event_handler_is_running: bool,
+}
+
+impl XPCClient {
+    pub fn connect(name: impl AsRef<CStr>) -> Self {
+        use block::ConcreteBlock;
+        use xpc_connection::xpc_object_to_message;
+        use xpc_connection_sys::xpc_connection_set_event_handler;
+        use xpc_connection_sys::xpc_connection_resume;
+
+        let name = name.as_ref();
+        let connection = unsafe {
+            xpc_connection_sys::xpc_connection_create_mach_service(name.as_ptr(), std::ptr::null_mut(), 0)
+        };
+
+        let (sender, receiver) = unbounded_channel();
+        let sender_clone = sender.clone();
+
+        let block = ConcreteBlock::new(move |event| {
+            let message = xpc_object_to_message(event);
+            sender_clone.unbounded_send(message).ok()
+        });
+
+        let block = block.copy();
+
+        unsafe {
+            xpc_connection_set_event_handler(connection, block.deref() as *const _ as *mut _);
+            xpc_connection_resume(connection);
+        }
+
+        Self {
+            connection,
+            receiver,
+            sender,
+            event_handler_is_running: true,
+        }
+    }
+
+    pub fn send_message(&self, message: Message) {
+        use xpc_connection::message_to_xpc_object;
+        use xpc_connection_sys::xpc_connection_send_message;
+        use xpc_connection_sys::xpc_release;
+
+        let xpc_object = message_to_xpc_object(message);
+        unsafe {
+            xpc_connection_send_message(self.connection, xpc_object);
+            xpc_release(xpc_object);
+        }
+    }
+}
+
+impl Drop for XPCClient {
+    fn drop(&mut self) {
+        use xpc_connection_sys::xpc_release;
+        use xpc_connection_sys::xpc_object_t;
+
+        unsafe { xpc_release(self.connection as xpc_object_t) };
+    }
+}
+
+impl Stream for XPCClient {
+    type Item = Message;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        match Stream::poll_next(Pin::new(&mut self.receiver), cx) {
+            Poll::Ready(Some(Message::Error(xpc_connection::MessageError::ConnectionInvalid))) => {
+                self.event_handler_is_running = false;
+                Poll::Ready(None)
+            }
+            v => v,
+        }
+    }
+}
+
+unsafe impl Send for XPCClient {}
 
 /// XPC-based implementation of DaemonInterface that sends method calls to the daemon over libxpc.
 pub struct XpcDaemonInterface;
@@ -25,33 +116,43 @@ impl XpcDaemonInterface {
 #[async_trait]
 impl DaemonInterface for XpcDaemonInterface {
     async fn print_version(&mut self) -> Result<()> {
+        // Build service name CString (trim trailing NUL from const)
+        let service_name = SERVICE_NAME.trim_end_matches('\0');
+        let mach_port_name = CString::new(service_name)?;
+
         // Open an XPC connection to the daemon service
-        let mut conn = XpcConnection::new(SERVICE_NAME);
-        let mut incoming = conn.connect();
+        let mut client = XPCClient::connect(&mach_port_name);
 
-        // Send a GetVersion request as a dictionary message
-        let mut dict = HashMap::new();
-        dict.insert(
-            GET_VERSION_METHOD.to_string(),
-            Message::String(String::new()),
-        );
-        conn.send_message(Message::Dictionary(dict));
-
-        // Wait for a single string reply (futures-preview StreamFuture returns (Option<Item>, Stream))
-        let (opt_msg, _) = match block_on(incoming.next()) {
-            Ok(pair) => pair,
-            Err(e) => {
-                eprintln!("Error reading XPC reply: {:?}", e);
-                return Ok(());
-            }
-        };
-        if let Some(Message::String(ver_raw)) = opt_msg {
-            // Trim the trailing NUL if present
-            let version = ver_raw.trim_end_matches('\0');
-            println!("Server version: {}", version);
-        } else {
-            eprintln!("Unexpected XPC reply for GetVersion");
+        // Send a GetVersion request as a dictionary message: { method: "GetVersion" }
+        {
+            let mut request = HashMap::new();
+            request.insert(
+                CString::new("method").unwrap(),
+                Message::String(CString::new(GET_VERSION_METHOD).unwrap()),
+            );
+            client.send_message(Message::Dictionary(request));
         }
+
+        // Await a single reply and print the version
+        match client.next().await {
+            Some(Message::Dictionary(map)) => {
+                if let Some(Message::String(ver)) = map.get(&CString::new("version").unwrap()) {
+                    println!("Server version: {}", ver.to_string_lossy());
+                } else if let Some(Message::String(ty)) = map.get(&CString::new("type").unwrap())
+                {
+                    println!("XPC replied with type: {}", ty.to_string_lossy());
+                } else {
+                    eprintln!("Unexpected XPC reply payload for GetVersion");
+                }
+            }
+            Some(other) => {
+                eprintln!("Unexpected XPC reply: {:?}", other);
+            }
+            None => {
+                eprintln!("No reply received from XPC daemon");
+            }
+        }
+
         Ok(())
     }