zbus/object_server/
mod.rs

1//! The object server API.
2
3use std::{collections::HashMap, marker::PhantomData, sync::Arc};
4use tracing::{Instrument, debug, instrument, trace, trace_span};
5
6use zbus_names::InterfaceName;
7use zvariant::{ObjectPath, Value};
8
9use crate::{
10    Connection, Error, Result,
11    async_lock::RwLock,
12    connection::WeakConnection,
13    fdo,
14    fdo::ObjectManager,
15    message::{Header, Message},
16};
17
18mod interface;
19pub(crate) use interface::ArcInterface;
20#[allow(deprecated)]
21pub use interface::DispatchResult;
22pub use interface::{DispatchResult2, Interface, InterfaceDeref, InterfaceDerefMut, InterfaceRef};
23
24mod signal_emitter;
25pub use signal_emitter::SignalEmitter;
26#[deprecated(since = "5.0.0", note = "Please use `SignalEmitter` instead.")]
27pub type SignalContext<'s> = SignalEmitter<'s>;
28
29mod dispatch_notifier;
30pub use dispatch_notifier::ResponseDispatchNotifier;
31
32mod node;
33pub(crate) use node::Node;
34
35/// An object server, holding server-side D-Bus objects & interfaces.
36///
37/// Object servers hold interfaces on various object paths, and expose them over D-Bus.
38///
39/// All object paths will have the standard interfaces implemented on your behalf, such as
40/// `org.freedesktop.DBus.Introspectable` or `org.freedesktop.DBus.Properties`.
41///
42/// # Example
43///
44/// This example exposes the `org.myiface.Example.Quit` method on the `/org/zbus/path`
45/// path.
46///
47/// ```no_run
48/// # use std::error::Error;
49/// use zbus::{Connection, interface};
50/// use event_listener::Event;
51/// # use async_io::block_on;
52///
53/// struct Example {
54///     // Interfaces are owned by the ObjectServer. They can have
55///     // `&mut self` methods.
56///     quit_event: Event,
57/// }
58///
59/// impl Example {
60///     fn new(quit_event: Event) -> Self {
61///         Self { quit_event }
62///     }
63/// }
64///
65/// #[interface(name = "org.myiface.Example")]
66/// impl Example {
67///     // This will be the "Quit" D-Bus method.
68///     async fn quit(&mut self) {
69///         self.quit_event.notify(1);
70///     }
71///
72///     // See `interface` documentation to learn
73///     // how to expose properties & signals as well.
74/// }
75///
76/// # block_on(async {
77/// let connection = Connection::session().await?;
78///
79/// let quit_event = Event::new();
80/// let quit_listener = quit_event.listen();
81/// let interface = Example::new(quit_event);
82/// connection
83///     .object_server()
84///     .at("/org/zbus/path", interface)
85///     .await?;
86///
87/// quit_listener.await;
88/// # Ok::<_, Box<dyn Error + Send + Sync>>(())
89/// # })?;
90/// # Ok::<_, Box<dyn Error + Send + Sync>>(())
91/// ```
92#[derive(Debug, Clone)]
93pub struct ObjectServer {
94    conn: WeakConnection,
95    root: Arc<RwLock<Node>>,
96}
97
98impl ObjectServer {
99    /// Create a new D-Bus `ObjectServer`.
100    pub(crate) fn new(conn: &Connection) -> Self {
101        Self {
102            conn: conn.into(),
103            root: Arc::new(RwLock::new(Node::new(
104                "/".try_into().expect("zvariant bug"),
105            ))),
106        }
107    }
108
109    pub(crate) fn root(&self) -> &RwLock<Node> {
110        &self.root
111    }
112
113    /// Register a D-Bus [`Interface`] at a given path (see the example above).
114    ///
115    /// Typically you'd want your interfaces to be registered immediately after the associated
116    /// connection is established and therefore use [`zbus::connection::Builder::serve_at`] instead.
117    /// However, there are situations where you'd need to register interfaces dynamically and that's
118    /// where this method becomes useful.
119    ///
120    /// If the interface already exists at this path, returns false.
121    pub async fn at<'p, P, I>(&self, path: P, iface: I) -> Result<bool>
122    where
123        I: Interface,
124        P: TryInto<ObjectPath<'p>>,
125        P::Error: Into<Error>,
126    {
127        self.add_arc_interface(path, I::name(), ArcInterface::new(iface))
128            .await
129    }
130
131    pub(crate) async fn add_arc_interface<'p, P>(
132        &self,
133        path: P,
134        name: InterfaceName<'static>,
135        arc_iface: ArcInterface,
136    ) -> Result<bool>
137    where
138        P: TryInto<ObjectPath<'p>>,
139        P::Error: Into<Error>,
140    {
141        let path = path.try_into().map_err(Into::into)?;
142        let mut root = self.root().write().await;
143        let (node, manager_path) = root.get_child_mut(&path, true);
144        let node = node.unwrap();
145        let added = node.add_arc_interface(name.clone(), arc_iface);
146        if added {
147            if name == ObjectManager::name() {
148                // Just added an object manager. Need to signal all managed objects under it.
149                let emitter = SignalEmitter::new(&self.connection(), path)?;
150                let objects = node.get_managed_objects(self, &self.connection()).await?;
151                for (path, owned_interfaces) in objects {
152                    let interfaces = owned_interfaces
153                        .iter()
154                        .map(|(i, props)| {
155                            let props = props
156                                .iter()
157                                .map(|(k, v)| Ok((k.as_str(), Value::try_from(v)?)))
158                                .collect::<Result<_>>();
159                            Ok((i.into(), props?))
160                        })
161                        .collect::<Result<_>>()?;
162                    ObjectManager::interfaces_added(&emitter, path.into(), interfaces).await?;
163                }
164            } else if let Some(manager_path) = manager_path {
165                let emitter = SignalEmitter::new(&self.connection(), manager_path.clone())?;
166                let mut interfaces = HashMap::new();
167                let owned_props = node
168                    .get_properties(self, &self.connection(), name.clone())
169                    .await?;
170                let props = owned_props
171                    .iter()
172                    .map(|(k, v)| Ok((k.as_str(), Value::try_from(v)?)))
173                    .collect::<Result<_>>()?;
174                interfaces.insert(name, props);
175
176                ObjectManager::interfaces_added(&emitter, path, interfaces).await?;
177            }
178        }
179
180        Ok(added)
181    }
182
183    /// Unregister a D-Bus [`Interface`] at a given path.
184    ///
185    /// If there are no more interfaces left at that path, destroys the object as well.
186    /// Returns whether the object was destroyed.
187    pub async fn remove<'p, I, P>(&self, path: P) -> Result<bool>
188    where
189        I: Interface,
190        P: TryInto<ObjectPath<'p>>,
191        P::Error: Into<Error>,
192    {
193        self.remove_named(path, I::name()).await
194    }
195
196    /// Unregister a D-Bus [`Interface`] at a given path, using its name.
197    ///
198    /// If there are no more interfaces left at that path, destroys the object as well.
199    /// Returns whether the object was destroyed.
200    pub async fn remove_named<'p, P>(
201        &self,
202        path: P,
203        interface_name: InterfaceName<'static>,
204    ) -> Result<bool>
205    where
206        P: TryInto<ObjectPath<'p>>,
207        P::Error: Into<Error>,
208    {
209        let path = path.try_into().map_err(Into::into)?;
210        let mut root = self.root.write().await;
211        let (node, manager_path) = root.get_child_mut(&path, false);
212        let node = node.ok_or(Error::InterfaceNotFound)?;
213        if !node.remove_interface(&interface_name) {
214            return Err(Error::InterfaceNotFound);
215        }
216        if let Some(manager_path) = manager_path {
217            let ctxt = SignalEmitter::new(&self.connection(), manager_path.clone())?;
218            ObjectManager::interfaces_removed(&ctxt, path.clone(), (&[interface_name]).into())
219                .await?;
220        }
221        if node.is_empty() {
222            let mut path_parts = path.rsplit('/').filter(|i| !i.is_empty());
223            let last_part = path_parts.next().unwrap();
224            let ppath = ObjectPath::from_string_unchecked(
225                path_parts.fold(String::new(), |a, p| format!("/{p}{a}")),
226            );
227            root.get_child_mut(&ppath, false)
228                .0
229                .unwrap()
230                .remove_node(last_part);
231            return Ok(true);
232        }
233        Ok(false)
234    }
235
236    /// Get the interface at the given path.
237    ///
238    /// # Errors
239    ///
240    /// If the interface is not registered at the given path, an `Error::InterfaceNotFound` error is
241    /// returned.
242    ///
243    /// # Examples
244    ///
245    /// The typical use of this is property changes outside of a dispatched handler:
246    ///
247    /// ```no_run
248    /// # use std::error::Error;
249    /// # use zbus::{Connection, interface};
250    /// # use async_io::block_on;
251    /// #
252    /// struct MyIface(u32);
253    ///
254    /// #[interface(name = "org.myiface.MyIface")]
255    /// impl MyIface {
256    ///      #[zbus(property)]
257    ///      async fn count(&self) -> u32 {
258    ///          self.0
259    ///      }
260    /// }
261    ///
262    /// # block_on(async {
263    /// # let connection = Connection::session().await?;
264    /// #
265    /// # let path = "/org/zbus/path";
266    /// # connection.object_server().at(path, MyIface(0)).await?;
267    /// let iface_ref = connection
268    ///     .object_server()
269    ///     .interface::<_, MyIface>(path).await?;
270    /// let mut iface = iface_ref.get_mut().await;
271    /// iface.0 = 42;
272    /// iface.count_changed(iface_ref.signal_emitter()).await?;
273    /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
274    /// # })?;
275    /// #
276    /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
277    /// ```
278    pub async fn interface<'p, P, I>(&self, path: P) -> Result<InterfaceRef<I>>
279    where
280        I: Interface,
281        P: TryInto<ObjectPath<'p>>,
282        P::Error: Into<Error>,
283    {
284        let path = path.try_into().map_err(Into::into)?;
285        let root = self.root().read().await;
286        let node = root.get_child(&path).ok_or(Error::InterfaceNotFound)?;
287
288        let lock = node
289            .interface_lock(I::name())
290            .ok_or(Error::InterfaceNotFound)?
291            .instance
292            .clone();
293
294        // Ensure what we return can later be dowcasted safely.
295        lock.read()
296            .await
297            .downcast_ref::<I>()
298            .ok_or(Error::InterfaceNotFound)?;
299
300        let conn = self.connection();
301        // SAFETY: We know that there is a valid path on the node as we already converted w/o error.
302        let emitter = SignalEmitter::new(&conn, path).unwrap().into_owned();
303
304        Ok(InterfaceRef {
305            emitter,
306            lock,
307            phantom: PhantomData,
308        })
309    }
310
311    async fn dispatch_call_to_iface(
312        &self,
313        iface: Arc<RwLock<dyn Interface>>,
314        connection: &Connection,
315        msg: &Message,
316        hdr: &Header<'_>,
317    ) -> fdo::Result<()> {
318        let member = hdr
319            .member()
320            .ok_or_else(|| fdo::Error::Failed("Missing member".into()))?;
321        let iface_name = hdr
322            .interface()
323            .ok_or_else(|| fdo::Error::Failed("Missing interface".into()))?;
324
325        trace!("acquiring read lock on interface `{}`", iface_name);
326        let read_lock = iface.read().await;
327        trace!("acquired read lock on interface `{}`", iface_name);
328        match read_lock.call(self, connection, msg, member.as_ref()) {
329            DispatchResult2::NotFound => {
330                return Err(fdo::Error::UnknownMethod(format!(
331                    "Unknown method '{member}'"
332                )));
333            }
334            DispatchResult2::Async(f) => {
335                return f.await;
336            }
337            DispatchResult2::RequiresMut => {}
338        }
339        drop(read_lock);
340        trace!("acquiring write lock on interface `{}`", iface_name);
341        let mut write_lock = iface.write().await;
342        trace!("acquired write lock on interface `{}`", iface_name);
343        match write_lock.call_mut(self, connection, msg, member.as_ref()) {
344            DispatchResult2::NotFound => {}
345            DispatchResult2::RequiresMut => {}
346            DispatchResult2::Async(f) => {
347                return f.await;
348            }
349        }
350        drop(write_lock);
351        Err(fdo::Error::UnknownMethod(format!(
352            "Unknown method '{member}'"
353        )))
354    }
355
356    async fn dispatch_method_call_try(
357        &self,
358        connection: &Connection,
359        msg: &Message,
360        hdr: &Header<'_>,
361    ) -> fdo::Result<()> {
362        let path = hdr
363            .path()
364            .ok_or_else(|| fdo::Error::Failed("Missing object path".into()))?;
365        let iface_name = hdr
366            .interface()
367            // TODO: In the absence of an INTERFACE field, if two or more interfaces on the same
368            // object have a method with the same name, it is undefined which of those
369            // methods will be invoked. Implementations may choose to either return an
370            // error, or deliver the message as though it had an arbitrary one of those
371            // interfaces.
372            .ok_or_else(|| fdo::Error::Failed("Missing interface".into()))?;
373        // Check that the message has a member before spawning.
374        // Note that an unknown member will still spawn a task. We should instead gather
375        // all the details for the call before spawning.
376        // See also https://github.com/z-galaxy/zbus/issues/674 for future of Interface.
377        let _ = hdr
378            .member()
379            .ok_or_else(|| fdo::Error::Failed("Missing member".into()))?;
380
381        // Ensure the root lock isn't held while dispatching the message. That
382        // way, the object server can be mutated during that time.
383        let (iface, with_spawn) = {
384            let root = self.root.read().await;
385
386            // D-Bus spec: org.freedesktop.DBus.Peer interface works on ANY path, even unregistered
387            // ones. See: https://dbus.freedesktop.org/doc/dbus-specification.html#standard-interfaces-peer
388            // Switch the path to "/" for Peer interface calls.
389            let path = if *iface_name == fdo::Peer::name() {
390                ObjectPath::from_static_str_unchecked("/")
391            } else {
392                path.clone()
393            };
394
395            let node = root
396                .get_child(&path)
397                .ok_or_else(|| fdo::Error::UnknownObject(format!("Unknown object '{path}'")))?;
398
399            let iface = node.interface_lock(iface_name.as_ref()).ok_or_else(|| {
400                fdo::Error::UnknownInterface(format!("Unknown interface '{iface_name}'"))
401            })?;
402            (iface.instance, iface.spawn_tasks_for_methods)
403        };
404
405        if with_spawn {
406            let executor = connection.executor().clone();
407            let task_name = format!("`{msg}` method dispatcher");
408            let connection = connection.clone();
409            let msg = msg.clone();
410            executor
411                .spawn(
412                    async move {
413                        let server = connection.object_server();
414                        let hdr = msg.header();
415                        if let Err(e) = server
416                            .dispatch_call_to_iface(iface, &connection, &msg, &hdr)
417                            .await
418                        {
419                            // When not spawning a task, this error is handled by the caller.
420                            debug!("Returning error: {}", e);
421                            if let Err(e) = connection.reply_dbus_error(&hdr, e).await {
422                                debug!(
423                                    "Error dispatching message. Message: {:?}, error: {:?}",
424                                    msg, e
425                                );
426                            }
427                        }
428                    }
429                    .instrument(trace_span!("{}", task_name)),
430                    &task_name,
431                )
432                .detach();
433            Ok(())
434        } else {
435            self.dispatch_call_to_iface(iface, connection, msg, hdr)
436                .await
437        }
438    }
439
440    /// Dispatch an incoming message to a registered interface.
441    ///
442    /// The object server will handle the message by:
443    ///
444    /// - looking up the called object path & interface,
445    ///
446    /// - calling the associated method if one exists,
447    ///
448    /// - returning a message (responding to the caller with either a return or error message) to
449    ///   the caller through the associated server connection.
450    ///
451    /// Returns an error if the message is malformed.
452    #[instrument(skip(self))]
453    pub(crate) async fn dispatch_call(&self, msg: &Message, hdr: &Header<'_>) -> Result<()> {
454        let conn = self.connection();
455
456        if let Err(e) = self.dispatch_method_call_try(&conn, msg, hdr).await {
457            debug!("Returning error: {}", e);
458            conn.reply_dbus_error(hdr, e).await?;
459        }
460        trace!("Handled: {}", msg);
461
462        Ok(())
463    }
464
465    pub(crate) fn connection(&self) -> Connection {
466        self.conn
467            .upgrade()
468            .expect("ObjectServer can't exist w/o an associated Connection")
469    }
470}
471
472#[cfg(feature = "blocking-api")]
473impl From<crate::blocking::ObjectServer> for ObjectServer {
474    fn from(server: crate::blocking::ObjectServer) -> Self {
475        server.into_inner()
476    }
477}
zbus/object_server/mod.rs

zbus/object_server/
mod.rs
