Full rewrite of the client without dispatchers

Cargo.toml

@@ -45,12 +45,14 @@ futures-util = { version = "0.3.31", default-features = false, features = [
 pin-project-lite = "0.2.15"
 rustix = { version = "0.38.42", features = ["fs", "net"] }
 thiserror = "2.0.7"
-tokio = { version = "1.42.0", features = ["net"] }
+tokio = { version = "1.42.0", features = ["net", "sync"] }
 tracing = "0.1.41"
 waynest-macros = { version = "0.0.2", path = "macros" }
 
 [dev-dependencies]
 anyhow = "1.0.94"
+async-stream = "0.3.6"
+tempfile = "3.14.0"
 tokio = { version = "1.41.1", features = ["rt-multi-thread", "macros"] }
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
 

examples/client.rs

@@ -3,56 +3,87 @@ use std::{borrow::Cow, collections::HashMap, os::unix::net::UnixStream, path::Pa
 use anyhow::Result;
 use futures_util::TryStreamExt;
 use waynest::{
-    client::{protocol::core::wayland::wl_display::WlDisplay, Dispatcher},
+    client::{protocol::core::wayland::wl_display::WlDisplay, Connection, ConnectionRef, Object},
     wire::{Message, ObjectId, Socket},
 };
 
-struct Client {
-    objects: HashMap<ObjectId, Arc<dyn Dispatcher>>,
-}
+use std::{
+    collections::HashMap,
+    sync::{Arc, Weak},
+};
 
-impl Client {
-    pub fn new() -> Result<Self> {
-        let mut objects: HashMap<ObjectId, Arc<dyn Dispatcher>> = HashMap::new();
+use futures_util::{Sink, SinkExt};
+use tokio::sync::{mpsc, Mutex};
 
-        objects.insert(ObjectId::DISPLAY, Arc::new(Display::new()));
+#[derive(Clone)]
+struct State {
+    tx: mpsc::Sender<Message>,
+    objects: HashMap<ObjectId, Arc<dyn Object>>,
+    next_id: u32,
+    _socket: Arc<Socket>,
+}
 
-        Ok(Self { objects })
+#[async_trait]
+impl ConnectionState for State {
+    async fn send(&self, value: Message) -> Result<(), Error> {
+        todo!()
     }
 
-    pub async fn _handle_message(&mut self, message: &Message) {
-        let dispatcher = self
-            .objects
-            .get(&message.object_id)
-            .expect("Invalid object id");
-
-        dispatcher.dispatch(message);
+    pub async fn register_object(
+        &self,
+        id: ObjectId,
+        object: Arc<dyn Object>,
+    ) -> Result<(), Error> {
+        let state = self.state.upgrade().ok_or(Error::Internal)?;
+        let mut state = state.lock().await;
+        state.objects.insert(id, object);
+        Ok(())
     }
 
-    pub fn display(&self) -> &Display {
-        let dispatcher = self
-            .objects
-            .get(&ObjectId::DISPLAY)
-            .expect("Invalid object id");
+    pub async fn unregister_object(&self, id: &ObjectId) -> Result<(), Error> {
+        let state = self.state.upgrade().ok_or(Error::Internal)?;
+        let mut state = state.lock().await;
+        state.objects.remove(id);
+        Ok(())
+    }
+}
 
-        dispatcher.downcast_ref().unwrap()
+impl State {
+    pub async fn next_id(&self) -> Result<ObjectId, Error> {
+        let state = self.state.upgrade().ok_or(Error::Internal)?;
+        let mut state = state.lock().await;
+        let id = state.next_id;
+        state.next_id += 1;
+        Ok(unsafe { ObjectId::from_raw(id) })
     }
 }
 
 #[derive(Clone)]
 struct Display {
-    // socket: Socket,
+    connection: ConnectionRef,
 }
 
 impl Display {
-    pub fn new() -> Self {
-        Self {}
+    pub fn new(connection: ConnectionRef) -> Self {
+        Self { connection }
     }
 }
 
-impl Dispatcher for Display {
-    fn dispatch(&self, _message: &Message) {
-        todo!()
+impl Object for Display {
+    fn interface(&self) -> &'static str {
+        Self::INTERFACE
+    }
+
+    fn version(&self) -> u32 {
+        Self::VERSION
+    }
+
+    fn id(&self) -> ObjectId {
+        ObjectId::DISPLAY
+    }
+
+    fn connection(&self) -> &ConnectionRef {
+        &self.connection
     }
 }
 
@@ -75,7 +106,7 @@ impl WlDisplay for Display {
 async fn main() -> Result<()> {
     tracing_subscriber::fmt::init();
 
-    let registry = unsafe { ObjectId::from_raw(2) };
+    // let registry = unsafe { ObjectId::from_raw(2) };
 
     let xdg_runtime_dir = std::env::var("XDG_RUNTIME_DIR")?;
     let wayland_connection = std::env::var("WAYLAND_DISPLAY")
@@ -86,17 +117,32 @@ async fn main() -> Result<()> {
 
     let mut socket = Socket::new(UnixStream::connect(socket_path)?)?;
 
-    let client = Client::new()?;
+    let connection = Connection::new(socket).await;
 
-    let display = client.display();
+    let display = Display::new(connection.downgrade());
 
-    display
-        .get_registry(&mut socket, ObjectId::DISPLAY, registry)
-        .await?;
+    // display
+    //     .connection()
+    //     .register_object(ObjectId::DISPLAY, Arc::new(display));
 
-    while let Some(message) = socket.try_next().await? {
-        dbg!(message);
-    }
+    // let client = Client::new()?;
+
+    // let display : &Display {
+    //     let dispatcher = self
+    //         .objects
+    //         .get(&ObjectId::DISPLAY)
+    //         .expect("Invalid object id");
+
+    //     dispatcher.downcast_ref().unwrap()
+    // }
+
+    // display
+    //     .get_registry(&mut socket, ObjectId::DISPLAY, registry)
+    //     .await?;
+
+    // while let Some(message) = socket.try_next().await? {
+    //     dbg!(message);
+    // }
 
     Ok(())
 }

gen/src/client.rs

@@ -31,22 +31,22 @@ pub fn generate_client_code(current: &[Pair], pairs: &[Pair]) -> TokenStream {
             let requests = write_requests(pairs, pair, interface);
             let events = write_events(pairs, pair, interface);
 
-            let imports = if requests.is_empty() {
-                quote! {}
-            } else {
-                quote! {use futures_util::SinkExt;}
-            };
+            // let imports = if requests.is_empty() {
+            //     quote! {}
+            // } else {
+            //     quote! {use futures_util::SinkExt;}
+            // };
 
             inner_modules.push(quote! {
                 #(#docs)*
                 #[allow(clippy::too_many_arguments)]
                 pub mod #module_name {
-                    #imports
+                    // #imports
 
                     #(#enums)*
 
                     #[doc = #trait_docs]
-                    pub trait #trait_name {
+                    pub trait #trait_name: crate::client::Object {
                         const INTERFACE: &'static str = #name;
                         const VERSION: u32 = #version;
 
@@ -100,11 +100,7 @@ fn write_requests(pairs: &[Pair], pair: &Pair, interface: &Interface) -> Vec<Tok
             request.name.to_snek_case()
         );
 
-        let mut args = vec![
-            quote! { &self },
-            quote! { socket: &mut crate::wire::Socket },
-            quote! { object_id: crate::wire::ObjectId },
-        ];
+        let mut args = vec![quote! { &self }];
 
         for arg in &request.args {
             let mut ty = arg.to_rust_type_token(arg.find_protocol(pairs).as_ref().unwrap_or(pair));
@@ -165,16 +161,17 @@ fn write_requests(pairs: &[Pair], pair: &Pair, interface: &Interface) -> Vec<Tok
         requests.push(quote! {
             #(#docs)*
             async fn #name(#(#args),*) -> crate::client::Result<()> {
+                let object_id = self.id();
+
                 tracing::debug!(#tracing_inner, object_id);
 
                 let (payload,fds) = crate::wire::PayloadBuilder::new()
                     #(#build_args)*
                     .build();
 
-                socket
-                    .send(crate::wire::Message::new(object_id, #opcode, payload, fds))
+                self.connection()
+                    .send_message(crate::wire::Message::new(object_id, #opcode, payload, fds))
                     .await
-                    .map_err(crate::client::Error::IoError)
             }
         });
     }
@@ -188,11 +185,7 @@ fn write_events(pairs: &[Pair], pair: &Pair, interface: &Interface) -> Vec<Token
     for request in &interface.events {
         let docs = description_to_docs(request.description.as_ref());
         let name = make_ident(request.name.to_snek_case());
-        let mut args = vec![
-            quote! {&self },
-            // quote! {object: &crate::server::Object},
-            // quote! {client: &mut crate::server::Client},
-        ];
+        let mut args = vec![quote! {&self }];
 
         for arg in &request.args {
             let mut ty = arg.to_rust_type_token(arg.find_protocol(pairs).as_ref().unwrap_or(pair));