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 ///
122 /// # Deadlocks
123 ///
124 /// It is fine to call this method from within an interface method (e.g. to register a child or
125 /// sibling object on demand), including from a `&mut self` method. There is however one
126 /// exception: registering an [`ObjectManager`] at an ancestor of the currently-executing
127 /// interface from within one of its `&mut self` methods will **deadlock**.
128 ///
129 /// This is because adding an `ObjectManager` reads the properties of every object under it (to
130 /// emit the initial `InterfacesAdded` signals), which requires a shared lock on each of those
131 /// interfaces — including the calling one, whose exclusive lock is held for the duration of the
132 /// `&mut self` method. Registering the `ObjectManager` up front (typically at connection
133 /// set-up), or from a `&self` method, avoids this.
134 pub async fn at<'p, P, I>(&self, path: P, iface: I) -> Result<bool>
135 where
136 I: Interface,
137 P: TryInto<ObjectPath<'p>>,
138 P::Error: Into<Error>,
139 {
140 self.add_arc_interface(path, I::name(), ArcInterface::new(iface))
141 .await
142 }
143
144 pub(crate) async fn add_arc_interface<'p, P>(
145 &self,
146 path: P,
147 name: InterfaceName<'static>,
148 arc_iface: ArcInterface,
149 ) -> Result<bool>
150 where
151 P: TryInto<ObjectPath<'p>>,
152 P::Error: Into<Error>,
153 {
154 let path = path.try_into().map_err(Into::into)?;
155 let mut root = self.root().write().await;
156 let (node, manager_path) = root.get_child_mut(&path, true);
157 let node = node.unwrap();
158 let added = node.add_arc_interface(name.clone(), arc_iface);
159 if added {
160 if name == ObjectManager::name() {
161 // Just added an object manager. Need to signal all managed objects under it.
162 let emitter = SignalEmitter::new(&self.connection(), path)?;
163 let objects = node.get_managed_objects(self, &self.connection()).await?;
164 for (path, owned_interfaces) in objects {
165 let interfaces = owned_interfaces
166 .iter()
167 .map(|(i, props)| {
168 let props = props
169 .iter()
170 .map(|(k, v)| Ok((k.as_str(), Value::try_from(v)?)))
171 .collect::<Result<_>>();
172 Ok((i.into(), props?))
173 })
174 .collect::<Result<_>>()?;
175 ObjectManager::interfaces_added(&emitter, path.into(), interfaces).await?;
176 }
177 } else if let Some(manager_path) = manager_path {
178 let emitter = SignalEmitter::new(&self.connection(), manager_path.clone())?;
179 let mut interfaces = HashMap::new();
180 let owned_props = node
181 .get_properties(self, &self.connection(), name.clone())
182 .await?;
183 let props = owned_props
184 .iter()
185 .map(|(k, v)| Ok((k.as_str(), Value::try_from(v)?)))
186 .collect::<Result<_>>()?;
187 interfaces.insert(name, props);
188
189 ObjectManager::interfaces_added(&emitter, path, interfaces).await?;
190 }
191 }
192
193 Ok(added)
194 }
195
196 /// Unregister a D-Bus [`Interface`] at a given path.
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<'p, I, P>(&self, path: P) -> Result<bool>
201 where
202 I: Interface,
203 P: TryInto<ObjectPath<'p>>,
204 P::Error: Into<Error>,
205 {
206 self.remove_named(path, I::name()).await
207 }
208
209 /// Unregister a D-Bus [`Interface`] at a given path, using its name.
210 ///
211 /// If there are no more interfaces left at that path, destroys the object as well.
212 /// Returns whether the object was destroyed.
213 pub async fn remove_named<'p, P>(
214 &self,
215 path: P,
216 interface_name: InterfaceName<'static>,
217 ) -> Result<bool>
218 where
219 P: TryInto<ObjectPath<'p>>,
220 P::Error: Into<Error>,
221 {
222 let path = path.try_into().map_err(Into::into)?;
223 let mut root = self.root.write().await;
224 let (node, manager_path) = root.get_child_mut(&path, false);
225 let node = node.ok_or(Error::InterfaceNotFound)?;
226 if !node.remove_interface(&interface_name) {
227 return Err(Error::InterfaceNotFound);
228 }
229 if let Some(manager_path) = manager_path {
230 let ctxt = SignalEmitter::new(&self.connection(), manager_path.clone())?;
231 ObjectManager::interfaces_removed(&ctxt, path.clone(), (&[interface_name]).into())
232 .await?;
233 }
234 if node.is_empty() {
235 let mut path_parts = path.rsplit('/').filter(|i| !i.is_empty());
236 let last_part = path_parts.next().unwrap();
237 let ppath = ObjectPath::from_string_unchecked(
238 path_parts.fold(String::new(), |a, p| format!("/{p}{a}")),
239 );
240 root.get_child_mut(&ppath, false)
241 .0
242 .unwrap()
243 .remove_node(last_part);
244 return Ok(true);
245 }
246 Ok(false)
247 }
248
249 /// Get the interface at the given path.
250 ///
251 /// # Errors
252 ///
253 /// If the interface is not registered at the given path, an `Error::InterfaceNotFound` error is
254 /// returned.
255 ///
256 /// # Examples
257 ///
258 /// The typical use of this is property changes outside of a dispatched handler:
259 ///
260 /// ```no_run
261 /// # use std::error::Error;
262 /// # use zbus::{Connection, interface};
263 /// # use async_io::block_on;
264 /// #
265 /// struct MyIface(u32);
266 ///
267 /// #[interface(name = "org.myiface.MyIface")]
268 /// impl MyIface {
269 /// #[zbus(property)]
270 /// async fn count(&self) -> u32 {
271 /// self.0
272 /// }
273 /// }
274 ///
275 /// # block_on(async {
276 /// # let connection = Connection::session().await?;
277 /// #
278 /// # let path = "/org/zbus/path";
279 /// # connection.object_server().at(path, MyIface(0)).await?;
280 /// let iface_ref = connection
281 /// .object_server()
282 /// .interface::<_, MyIface>(path).await?;
283 /// let mut iface = iface_ref.get_mut().await;
284 /// iface.0 = 42;
285 /// iface.count_changed(iface_ref.signal_emitter()).await?;
286 /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
287 /// # })?;
288 /// #
289 /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
290 /// ```
291 pub async fn interface<'p, P, I>(&self, path: P) -> Result<InterfaceRef<I>>
292 where
293 I: Interface,
294 P: TryInto<ObjectPath<'p>>,
295 P::Error: Into<Error>,
296 {
297 let path = path.try_into().map_err(Into::into)?;
298 let root = self.root().read().await;
299 let node = root.get_child(&path).ok_or(Error::InterfaceNotFound)?;
300
301 let lock = node
302 .interface_lock(I::name())
303 .ok_or(Error::InterfaceNotFound)?
304 .instance
305 .clone();
306
307 // Ensure what we return can later be dowcasted safely.
308 lock.read()
309 .await
310 .downcast_ref::<I>()
311 .ok_or(Error::InterfaceNotFound)?;
312
313 let conn = self.connection();
314 // SAFETY: We know that there is a valid path on the node as we already converted w/o error.
315 let emitter = SignalEmitter::new(&conn, path).unwrap().into_owned();
316
317 Ok(InterfaceRef {
318 emitter,
319 lock,
320 phantom: PhantomData,
321 })
322 }
323
324 async fn dispatch_call_to_iface(
325 &self,
326 iface: Arc<RwLock<dyn Interface>>,
327 connection: &Connection,
328 msg: &Message,
329 hdr: &Header<'_>,
330 ) -> fdo::Result<()> {
331 let member = hdr
332 .member()
333 .ok_or_else(|| fdo::Error::Failed("Missing member".into()))?;
334 let iface_name = hdr
335 .interface()
336 .ok_or_else(|| fdo::Error::Failed("Missing interface".into()))?;
337
338 trace!("acquiring read lock on interface `{}`", iface_name);
339 let read_lock = iface.read().await;
340 trace!("acquired read lock on interface `{}`", iface_name);
341 match read_lock.call(self, connection, msg, member.as_ref()) {
342 DispatchResult2::NotFound => {
343 return Err(fdo::Error::UnknownMethod(format!(
344 "Unknown method '{member}'"
345 )));
346 }
347 DispatchResult2::Async(f) => {
348 return f.await;
349 }
350 DispatchResult2::RequiresMut => {}
351 }
352 drop(read_lock);
353 trace!("acquiring write lock on interface `{}`", iface_name);
354 let mut write_lock = iface.write().await;
355 trace!("acquired write lock on interface `{}`", iface_name);
356 match write_lock.call_mut(self, connection, msg, member.as_ref()) {
357 DispatchResult2::NotFound => {}
358 DispatchResult2::RequiresMut => {}
359 DispatchResult2::Async(f) => {
360 return f.await;
361 }
362 }
363 drop(write_lock);
364 Err(fdo::Error::UnknownMethod(format!(
365 "Unknown method '{member}'"
366 )))
367 }
368
369 async fn dispatch_method_call_try(
370 &self,
371 connection: &Connection,
372 msg: &Message,
373 hdr: &Header<'_>,
374 ) -> fdo::Result<()> {
375 let path = hdr
376 .path()
377 .ok_or_else(|| fdo::Error::Failed("Missing object path".into()))?;
378 let iface_name = hdr
379 .interface()
380 // TODO: In the absence of an INTERFACE field, if two or more interfaces on the same
381 // object have a method with the same name, it is undefined which of those
382 // methods will be invoked. Implementations may choose to either return an
383 // error, or deliver the message as though it had an arbitrary one of those
384 // interfaces.
385 .ok_or_else(|| fdo::Error::Failed("Missing interface".into()))?;
386 // Check that the message has a member before spawning.
387 // Note that an unknown member will still spawn a task. We should instead gather
388 // all the details for the call before spawning.
389 // See also https://github.com/z-galaxy/zbus/issues/674 for future of Interface.
390 let _ = hdr
391 .member()
392 .ok_or_else(|| fdo::Error::Failed("Missing member".into()))?;
393
394 // Ensure the root lock isn't held while dispatching the message. That
395 // way, the object server can be mutated during that time.
396 let (iface, with_spawn) = {
397 let root = self.root.read().await;
398
399 // D-Bus spec: org.freedesktop.DBus.Peer interface works on ANY path, even unregistered
400 // ones. See: https://dbus.freedesktop.org/doc/dbus-specification.html#standard-interfaces-peer
401 // Switch the path to "/" for Peer interface calls.
402 let path = if *iface_name == fdo::Peer::name() {
403 ObjectPath::from_static_str_unchecked("/")
404 } else {
405 path.clone()
406 };
407
408 let node = root
409 .get_child(&path)
410 .ok_or_else(|| fdo::Error::UnknownObject(format!("Unknown object '{path}'")))?;
411
412 let iface = node.interface_lock(iface_name.as_ref()).ok_or_else(|| {
413 fdo::Error::UnknownInterface(format!("Unknown interface '{iface_name}'"))
414 })?;
415 (iface.instance, iface.spawn_tasks_for_methods)
416 };
417
418 if with_spawn {
419 let executor = connection.executor().clone();
420 let task_name = format!("`{msg}` method dispatcher");
421 let connection = connection.clone();
422 let msg = msg.clone();
423 executor
424 .spawn(
425 async move {
426 let server = connection.object_server();
427 let hdr = msg.header();
428 if let Err(e) = server
429 .dispatch_call_to_iface(iface, &connection, &msg, &hdr)
430 .await
431 {
432 // When not spawning a task, this error is handled by the caller.
433 debug!("Returning error: {}", e);
434 if let Err(e) = connection.reply_dbus_error(&hdr, e).await {
435 debug!(
436 "Error dispatching message. Message: {:?}, error: {:?}",
437 msg, e
438 );
439 }
440 }
441 }
442 .instrument(trace_span!("{}", task_name)),
443 &task_name,
444 )
445 .detach();
446 Ok(())
447 } else {
448 self.dispatch_call_to_iface(iface, connection, msg, hdr)
449 .await
450 }
451 }
452
453 /// Dispatch an incoming message to a registered interface.
454 ///
455 /// The object server will handle the message by:
456 ///
457 /// - looking up the called object path & interface,
458 ///
459 /// - calling the associated method if one exists,
460 ///
461 /// - returning a message (responding to the caller with either a return or error message) to
462 /// the caller through the associated server connection.
463 ///
464 /// Returns an error if the message is malformed.
465 #[instrument(skip(self))]
466 pub(crate) async fn dispatch_call(&self, msg: &Message, hdr: &Header<'_>) -> Result<()> {
467 let conn = self.connection();
468
469 if let Err(e) = self.dispatch_method_call_try(&conn, msg, hdr).await {
470 debug!("Returning error: {}", e);
471 conn.reply_dbus_error(hdr, e).await?;
472 }
473 trace!("Handled: {}", msg);
474
475 Ok(())
476 }
477
478 pub(crate) fn connection(&self) -> Connection {
479 self.conn
480 .upgrade()
481 .expect("ObjectServer can't exist w/o an associated Connection")
482 }
483}
484
485#[cfg(feature = "blocking-api")]
486impl From<crate::blocking::ObjectServer> for ObjectServer {
487 fn from(server: crate::blocking::ObjectServer) -> Self {
488 server.into_inner()
489 }
490}