use bytes::BufMut; use serde::Serialize; use smallvec::SmallVec; use smol::io::AsyncReadExt; use smol::io::AsyncWriteExt; use smol::lock::Mutex; use smol::net::unix::UnixStream; use std::sync::{Arc, Weak}; use crate::{ gen_id, ipc::{self, Serial}, packet_client::{self, HandsfreeParams, PacketClient}, packet_server::{self, PacketServer}, util::notifier::Notifier, }; pub struct QueuedPacket { notifier: Notifier, serial: Serial, packet: Option, } gen_id!( QueuedPacketVec, QueuedPacket, QueuedPacketCell, QueuedPacketHandle ); #[derive(Debug, Serialize, Clone)] pub struct AuthInfo { pub runtime: String, } type SignalFunc = Box bool + Send>; pub struct WayVRClient { receiver: ReceiverMutex, sender: SenderMutex, cancel_tx: async_channel::Sender<()>, exiting: bool, queued_packets: QueuedPacketVec, pub auth: Option, pub on_signal: Option, } pub async fn send_packet(sender: &SenderMutex, data: &[u8]) -> anyhow::Result<()> { let mut bytes = bytes::BytesMut::new(); // packet size bytes.put_u32(data.len() as u32); // packet data bytes.put_slice(data); sender.lock().await.write_all(&bytes).await?; Ok(()) } pub type WayVRClientMutex = Arc>; pub type WayVRClientWeak = Weak>; type ReceiverMutex = Arc>; type SenderMutex = Arc>; async fn client_runner(client: WayVRClientMutex) -> anyhow::Result<()> { loop { WayVRClient::tick(client.clone()).await?; } } type Payload = SmallVec<[u8; 64]>; async fn read_payload( conn: &mut smol::net::unix::UnixStream, size: u32, ) -> anyhow::Result { let mut payload = Payload::new(); payload.resize(size as usize, 0); conn.read_exact(&mut payload).await?; Ok(payload) } macro_rules! bail_unexpected_response { () => { anyhow::bail!("unexpected response"); }; } // Send and wait for a response from the server macro_rules! send_and_wait { ($client_mtx:expr, $serial:expr, $packet_to_send:expr, $expected_response_type:ident) => {{ let result = WayVRClient::queue_wait_packet($client_mtx, $serial, &ipc::data_encode($packet_to_send)) .await?; if let PacketServer::$expected_response_type(_, res) = result { res } else { bail_unexpected_response!(); } }}; } // Send without expecting any response macro_rules! send_only { ($client_mtx:expr, $packet_to_send:expr) => {{ WayVRClient::send_payload($client_mtx, &ipc::data_encode($packet_to_send)).await?; }}; } impl WayVRClient { pub fn set_signal_handler(&mut self, on_signal: SignalFunc) { self.on_signal = Some(on_signal); } pub async fn new(client_name: &str) -> anyhow::Result { let printname = "/tmp/wayvr_ipc.sock"; let stream = match UnixStream::connect(printname).await { Ok(c) => c, Err(e) => { anyhow::bail!("Failed to connect to the WayVR IPC: {}", e) } }; let receiver = Arc::new(Mutex::new(stream.clone())); let sender = Arc::new(Mutex::new(stream)); let (cancel_tx, cancel_rx) = async_channel::bounded(1); let client = Arc::new(Mutex::new(Self { receiver, sender: sender.clone(), cancel_tx, exiting: false, queued_packets: QueuedPacketVec::new(), auth: None, on_signal: None, })); WayVRClient::start_runner(client.clone(), cancel_rx); // Send handshake to the server send_packet( &sender, &ipc::data_encode(&PacketClient::Handshake(packet_client::Handshake { client_name: String::from(client_name), magic: String::from(ipc::CONNECTION_MAGIC), protocol_version: ipc::PROTOCOL_VERSION, })), ) .await?; Ok(client) } fn start_runner(client: WayVRClientMutex, cancel_rx: async_channel::Receiver<()>) { smol::spawn(async move { let cancel_fut = async { cancel_rx .recv() .await .map_err(|_| anyhow::anyhow!("cancelled")) }; let result = smol::future::race(client_runner(client.clone()), cancel_fut).await; match result { Ok(()) => { log::info!("IPC Runner completed"); } Err(e) => { if e.to_string() == "cancelled" { log::info!("Exiting IPC runner gracefully"); } else { log::info!("IPC Runner failed: {:?}", e); } } } }) .detach(); } async fn tick(client_mtx: WayVRClientMutex) -> anyhow::Result<()> { let receiver = { let client = client_mtx.lock().await; client.receiver.clone() }; // read packet let packet = { let mut receiver = receiver.lock().await; let mut size_buf = [0u8; 4]; receiver.read_exact(&mut size_buf).await?; let packet_size = u32::from_be_bytes(size_buf); log::trace!("packet size {}", packet_size); if packet_size > 128 * 1024 { anyhow::bail!("packet size too large (> 128 KiB)"); } let payload = read_payload(&mut receiver, packet_size).await?; let packet: PacketServer = ipc::data_decode(&payload)?; packet }; { let mut client = client_mtx.lock().await; if let PacketServer::HandshakeSuccess(success) = &packet { if client.auth.is_some() { anyhow::bail!("Got handshake response twice"); } client.auth = Some(AuthInfo { runtime: success.runtime.clone(), }); log::info!( "Authenticated. Server runtime name: \"{}\"", success.runtime ); } if let PacketServer::Disconnect(disconnect) = &packet { anyhow::bail!("Server disconnected us. Reason: {}", disconnect.reason); } if client.auth.is_none() { anyhow::bail!( "Server tried to send us a packet which is not a HandshakeSuccess or Disconnect" ); } if let PacketServer::WvrStateChanged(_) = &packet && let Some(on_signal) = &mut client.on_signal && (*on_signal)(&packet) { // Signal consumed return Ok(()); } // queue packet to read if it contains a serial response if let Some(serial) = packet.serial() { for qpacket in &mut client.queued_packets.vec { let Some(qpacket) = qpacket else { continue; }; let qpacket = &mut qpacket.obj; if qpacket.serial != *serial { continue; //skip } // found response serial, fill it and notify the receiver qpacket.packet = Some(packet); let notifier = qpacket.notifier.clone(); drop(client); notifier.notify(); break; } } } Ok(()) } // Send packet without feedback async fn send_payload(client_mtx: WayVRClientMutex, payload: &[u8]) -> anyhow::Result<()> { let client = client_mtx.lock().await; let sender = client.sender.clone(); drop(client); send_packet(&sender, payload).await?; Ok(()) } async fn queue_wait_packet( client_mtx: WayVRClientMutex, serial: Serial, payload: &[u8], ) -> anyhow::Result { let notifier = Notifier::new(); // Send packet to the server let queued_packet_handle = { let mut client = client_mtx.lock().await; let handle = client.queued_packets.add(QueuedPacket { notifier: notifier.clone(), packet: None, // will be filled after notify serial, }); let sender = client.sender.clone(); drop(client); send_packet(&sender, payload).await?; handle }; // Wait for response message notifier.wait().await; // Fetch response packet { let mut client = client_mtx.lock().await; let cell = client .queued_packets .get_mut(&queued_packet_handle) .ok_or(anyhow::anyhow!( "missing packet cell, this shouldn't happen" ))?; let Some(packet) = cell.packet.take() else { anyhow::bail!("packet is None, this shouldn't happen"); }; client.queued_packets.remove(&queued_packet_handle); Ok(packet) } } pub async fn fn_wvr_window_list( client: WayVRClientMutex, serial: Serial, ) -> anyhow::Result>> { Ok( send_and_wait!( client, serial, &PacketClient::WvrWindowList(serial), WvrWindowListResponse ) .map(|res| res.list), ) } pub async fn fn_wvr_window_set_visible( client: WayVRClientMutex, handle: packet_server::WvrWindowHandle, visible: bool, ) -> anyhow::Result<()> { send_only!(client, &PacketClient::WvrWindowSetVisible(handle, visible)); Ok(()) } pub async fn fn_wvr_process_list( client: WayVRClientMutex, serial: Serial, ) -> anyhow::Result> { Ok( send_and_wait!( client, serial, &PacketClient::WvrProcessList(serial), WvrProcessListResponse ) .list, ) } pub async fn fn_wvr_process_get( client: WayVRClientMutex, serial: Serial, handle: packet_server::WvrProcessHandle, ) -> anyhow::Result> { Ok(send_and_wait!( client, serial, &PacketClient::WvrProcessGet(serial, handle), WvrProcessGetResponse )) } pub async fn fn_wvr_process_terminate( client: WayVRClientMutex, handle: packet_server::WvrProcessHandle, ) -> anyhow::Result<()> { send_only!(client, &PacketClient::WvrProcessTerminate(handle)); Ok(()) } pub async fn fn_wvr_process_launch( client: WayVRClientMutex, serial: Serial, params: packet_client::WvrProcessLaunchParams, ) -> anyhow::Result { send_and_wait!( client, serial, &PacketClient::WvrProcessLaunch(serial, params), WvrProcessLaunchResponse ) .map_err(|e| anyhow::anyhow!("{}", e)) } pub async fn fn_wlx_input_state( client: WayVRClientMutex, serial: Serial, ) -> anyhow::Result { Ok(send_and_wait!( client, serial, &PacketClient::WlxInputState(serial), WlxInputStateResponse )) } pub async fn fn_wlx_device_haptics( client: WayVRClientMutex, device: usize, params: packet_client::WlxHapticsParams, ) -> anyhow::Result<()> { send_only!(client, &PacketClient::WlxDeviceHaptics(device, params)); Ok(()) } pub async fn fn_wlx_show_hide(client: WayVRClientMutex) -> anyhow::Result<()> { send_only!(client, &PacketClient::WlxShowHide); Ok(()) } pub async fn fn_wlx_switch_set( client: WayVRClientMutex, set: Option, ) -> anyhow::Result<()> { send_only!(client, &PacketClient::WlxSwitchSet(set)); Ok(()) } pub async fn fn_wlx_handsfree( client: WayVRClientMutex, params: HandsfreeParams, ) -> anyhow::Result<()> { send_only!(client, &PacketClient::WlxHandsfree(params)); Ok(()) } pub async fn fn_wlx_modify_panel( client: WayVRClientMutex, params: packet_client::WlxModifyPanelParams, ) -> anyhow::Result<()> { send_only!(client, &PacketClient::WlxModifyPanel(params)); Ok(()) } } impl Drop for WayVRClient { fn drop(&mut self) { self.exiting = true; let _ = self.cancel_tx.try_send(()); } }