zbus/connection/builder.rs
1use async_broadcast::Receiver as ActiveReceiver;
2#[cfg(feature = "async-io")]
3use async_io::Async;
4use enumflags2::BitFlags;
5use event_listener::Event;
6#[cfg(not(feature = "tokio"))]
7use std::net::TcpStream;
8#[cfg(all(unix, not(feature = "tokio")))]
9use std::os::unix::net::UnixStream;
10use std::{
11 collections::{HashMap, HashSet},
12 vec,
13};
14#[cfg(feature = "tokio")]
15use tokio::net::TcpStream;
16#[cfg(all(unix, feature = "tokio"))]
17use tokio::net::UnixStream;
18#[cfg(feature = "tokio-vsock")]
19use tokio_vsock::VsockStream;
20#[cfg(all(windows, not(feature = "tokio")))]
21use uds_windows::UnixStream;
22#[cfg(all(feature = "vsock", not(feature = "tokio-vsock")))]
23use vsock::VsockStream;
24
25// Feature-independent stream types for the `async_io_*_stream` builders: these always take the
26// blocking/`async-io` stream, so enabling `tokio` elsewhere can't change what they accept.
27#[cfg(feature = "async-io")]
28use std::net::TcpStream as AsyncIoTcpStream;
29#[cfg(all(unix, feature = "async-io"))]
30use std::os::unix::net::UnixStream as AsyncIoUnixStream;
31#[cfg(all(windows, feature = "async-io"))]
32use uds_windows::UnixStream as AsyncIoUnixStream;
33
34use zvariant::ObjectPath;
35
36#[cfg(feature = "bus-impl")]
37use crate::MessageStream;
38use crate::{
39 Connection, Error, Executor, Guid, OwnedGuid, Result,
40 address::{self, Address},
41 fdo::RequestNameFlags,
42 message::Message,
43 names::{InterfaceName, WellKnownName},
44 object_server::{ArcInterface, Interface},
45};
46
47use super::{
48 handshake::{AuthMechanism, Authenticated},
49 socket::{BoxedSplit, ReadHalf, Split, WriteHalf},
50};
51
52const DEFAULT_MAX_QUEUED: usize = 64;
53
54#[derive(Debug)]
55enum Target {
56 #[cfg(all(unix, feature = "tokio"))]
57 TokioUnixStream(tokio::net::UnixStream),
58 #[cfg(all(any(unix, windows), feature = "async-io"))]
59 AsyncIoUnixStream(AsyncIoUnixStream),
60 #[cfg(feature = "tokio")]
61 TokioTcpStream(tokio::net::TcpStream),
62 #[cfg(feature = "async-io")]
63 AsyncIoTcpStream(AsyncIoTcpStream),
64 #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
65 VsockStream(VsockStream),
66 Address(Address),
67 Socket(Split<Box<dyn ReadHalf>, Box<dyn WriteHalf>>),
68 AuthenticatedSocket(Split<Box<dyn ReadHalf>, Box<dyn WriteHalf>>),
69}
70
71type Interfaces<'a> = HashMap<ObjectPath<'a>, HashMap<InterfaceName<'static>, ArcInterface>>;
72
73/// A builder for [`zbus::Connection`].
74///
75/// The builder allows setting the flags [`RequestNameFlags::AllowReplacement`] and
76/// [`RequestNameFlags::ReplaceExisting`] when requesting names, but the flag
77/// [`RequestNameFlags::DoNotQueue`] will always be enabled. The reasons are:
78///
79/// 1. There is no indication given to the caller of [`Self::build`] that the name(s) request was
80/// enqueued and that the requested name might not be available right after building.
81///
82/// 2. The name may be acquired in between the time the name is requested and the
83/// [`crate::fdo::NameAcquiredStream`] is constructed. As a result the service can miss the
84/// [`crate::fdo::NameAcquired`] signal.
85#[derive(Debug)]
86#[must_use]
87pub struct Builder<'a> {
88 target: Option<Target>,
89 max_queued: Option<usize>,
90 // This is only set for p2p server case or pre-authenticated sockets.
91 guid: Option<Guid<'a>>,
92 #[cfg(feature = "p2p")]
93 p2p: bool,
94 internal_executor: bool,
95 interfaces: Interfaces<'a>,
96 names: HashSet<WellKnownName<'a>>,
97 auth_mechanism: Option<AuthMechanism>,
98 #[cfg(feature = "bus-impl")]
99 unique_name: Option<crate::names::UniqueName<'a>>,
100 request_name_flags: BitFlags<RequestNameFlags>,
101 method_timeout: Option<std::time::Duration>,
102 user_id: Option<u32>,
103}
104
105impl<'a> Builder<'a> {
106 /// Create a builder for the session/user message bus connection.
107 pub fn session() -> Result<Self> {
108 Ok(Self::new(Target::Address(Address::session()?)))
109 }
110
111 /// Create a builder for the system-wide message bus connection.
112 pub fn system() -> Result<Self> {
113 Ok(Self::new(Target::Address(Address::system()?)))
114 }
115
116 /// Create a builder for an IBus connection.
117 ///
118 /// IBus (Intelligent Input Bus) is an input method framework. This method creates a builder
119 /// that will query the IBus daemon for its D-Bus address using the `ibus address` command.
120 ///
121 /// # Platform Support
122 ///
123 /// This method is available on Unix-like systems where IBus is installed.
124 ///
125 /// # Errors
126 ///
127 /// Returns an error if:
128 /// - The `ibus` command is not found or fails to execute
129 /// - The IBus daemon is not running
130 /// - The command output cannot be parsed as a valid D-Bus address
131 ///
132 /// # Example
133 ///
134 /// ```no_run
135 /// # use std::error::Error;
136 /// # use zbus::connection::Builder;
137 /// # use zbus::block_on;
138 /// #
139 /// # block_on(async {
140 /// let conn = Builder::ibus()?
141 /// .build()
142 /// .await?;
143 ///
144 /// // Use the connection to interact with IBus services
145 /// # drop(conn);
146 /// # Ok::<(), zbus::Error>(())
147 /// # }).unwrap();
148 /// #
149 /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
150 /// ```
151 #[cfg(unix)]
152 pub fn ibus() -> Result<Self> {
153 use crate::address::transport::{Ibus, Transport};
154 Ok(Self::new(Target::Address(Address::from(Transport::Ibus(
155 Ibus::new(),
156 )))))
157 }
158
159 /// Create a builder for a connection that will use the given [D-Bus bus address].
160 ///
161 /// # Example
162 ///
163 /// Here is an example of connecting to an IBus service:
164 ///
165 /// ```no_run
166 /// # use std::error::Error;
167 /// # use zbus::connection::Builder;
168 /// # use zbus::block_on;
169 /// #
170 /// # block_on(async {
171 /// let addr = "unix:\
172 /// path=/home/zeenix/.cache/ibus/dbus-ET0Xzrk9,\
173 /// guid=fdd08e811a6c7ebe1fef0d9e647230da";
174 /// let conn = Builder::address(addr)?
175 /// .build()
176 /// .await?;
177 ///
178 /// // Do something useful with `conn`..
179 /// # drop(conn);
180 /// # Ok::<(), zbus::Error>(())
181 /// # }).unwrap();
182 /// #
183 /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
184 /// ```
185 ///
186 /// **Note:** The IBus address is different for each session. You can find the address for your
187 /// current session using `ibus address` command. For a more convenient way to connect to IBus,
188 /// see [`Builder::ibus`].
189 ///
190 /// [D-Bus bus address]: https://dbus.freedesktop.org/doc/dbus-specification.html#addresses
191 pub fn address<A>(address: A) -> Result<Self>
192 where
193 A: TryInto<Address>,
194 A::Error: Into<Error>,
195 {
196 Ok(Self::new(Target::Address(
197 address.try_into().map_err(Into::into)?,
198 )))
199 }
200
201 /// Create a builder for a connection that will use the given unix stream.
202 ///
203 /// The stream is a [`std::os::unix::net::UnixStream`] (or [`uds_windows::UnixStream`] on
204 /// Windows).
205 ///
206 /// [`uds_windows::UnixStream`]: https://docs.rs/uds_windows/latest/uds_windows/struct.UnixStream.html
207 #[cfg(all(any(unix, windows), feature = "async-io"))]
208 pub fn async_io_unix_stream(stream: AsyncIoUnixStream) -> Self {
209 Self::new(Target::AsyncIoUnixStream(stream))
210 }
211
212 /// Create a builder for a connection that will use the given unix stream.
213 ///
214 /// This method expects a
215 /// [`tokio::net::UnixStream`](https://docs.rs/tokio/latest/tokio/net/struct.UnixStream.html).
216 /// Without the `tokio` feature it accepts a [`std::os::unix::net::UnixStream`] instead, but
217 /// that form is deprecated in favor of
218 /// [`async_io_unix_stream`](Self::async_io_unix_stream).
219 ///
220 /// Since tokio currently [does not support Unix domain sockets][tuds] on Windows, this method
221 /// is not available when the `tokio` feature is enabled and building for Windows target.
222 ///
223 /// [tuds]: https://github.com/tokio-rs/tokio/issues/2201
224 #[cfg_attr(
225 not(feature = "tokio"),
226 deprecated(
227 since = "5.19.0",
228 note = "Use `async_io_unix_stream` to avoid a build failure if the `tokio` feature gets enabled"
229 )
230 )]
231 #[cfg(any(unix, not(feature = "tokio")))]
232 pub fn unix_stream(stream: UnixStream) -> Self {
233 #[cfg(not(feature = "tokio"))]
234 {
235 Self::new(Target::AsyncIoUnixStream(stream))
236 }
237 #[cfg(feature = "tokio")]
238 {
239 Self::new(Target::TokioUnixStream(stream))
240 }
241 }
242
243 /// Create a builder for a connection that will use the given TCP stream.
244 ///
245 /// The stream is a [`std::net::TcpStream`].
246 #[cfg(feature = "async-io")]
247 pub fn async_io_tcp_stream(stream: AsyncIoTcpStream) -> Self {
248 Self::new(Target::AsyncIoTcpStream(stream))
249 }
250
251 /// Create a builder for a connection that will use the given TCP stream.
252 ///
253 /// This method expects a
254 /// [`tokio::net::TcpStream`](https://docs.rs/tokio/latest/tokio/net/struct.TcpStream.html).
255 /// Without the `tokio` feature it accepts a [`std::net::TcpStream`] instead, but that form is
256 /// deprecated in favor of [`async_io_tcp_stream`](Self::async_io_tcp_stream).
257 #[cfg_attr(
258 not(feature = "tokio"),
259 deprecated(
260 since = "5.19.0",
261 note = "Use `async_io_tcp_stream` to avoid a build failure if the `tokio` feature gets enabled"
262 )
263 )]
264 pub fn tcp_stream(stream: TcpStream) -> Self {
265 #[cfg(not(feature = "tokio"))]
266 {
267 Self::new(Target::AsyncIoTcpStream(stream))
268 }
269 #[cfg(feature = "tokio")]
270 {
271 Self::new(Target::TokioTcpStream(stream))
272 }
273 }
274
275 /// Create a builder for a connection that will use the given VSOCK stream.
276 ///
277 /// This method is only available when either `vsock` or `tokio-vsock` feature is enabled. The
278 /// type of `stream` is `vsock::VsockStream` with `vsock` feature and `tokio_vsock::VsockStream`
279 /// with `tokio-vsock` feature.
280 #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
281 pub fn vsock_stream(stream: VsockStream) -> Self {
282 Self::new(Target::VsockStream(stream))
283 }
284
285 /// Create a builder for a connection that will use the given socket.
286 pub fn socket<S: Into<BoxedSplit>>(socket: S) -> Self {
287 Self::new(Target::Socket(socket.into()))
288 }
289
290 /// Create a builder for a connection that will use the given pre-authenticated socket.
291 ///
292 /// This is similar to [`Builder::socket`], except that the socket is either already
293 /// authenticated or does not require authentication.
294 pub fn authenticated_socket<S, G>(socket: S, guid: G) -> Result<Self>
295 where
296 S: Into<BoxedSplit>,
297 G: TryInto<Guid<'a>>,
298 G::Error: Into<Error>,
299 {
300 let mut builder = Self::new(Target::AuthenticatedSocket(socket.into()));
301 builder.guid = Some(guid.try_into().map_err(Into::into)?);
302
303 Ok(builder)
304 }
305
306 /// Specify the mechanism to use during authentication.
307 pub fn auth_mechanism(mut self, auth_mechanism: AuthMechanism) -> Self {
308 self.auth_mechanism = Some(auth_mechanism);
309
310 self
311 }
312
313 /// Specify the user id during authentication.
314 ///
315 /// This can be useful when using [`AuthMechanism::External`] with `socat`
316 /// to avoid the host decide what uid to use and instead provide one
317 /// known to have access rights.
318 #[cfg(unix)]
319 pub fn user_id(mut self, id: u32) -> Self {
320 self.user_id = Some(id);
321
322 self
323 }
324
325 /// The to-be-created connection will be a peer-to-peer connection.
326 ///
327 /// This method is only available when the `p2p` feature is enabled.
328 #[cfg(feature = "p2p")]
329 pub fn p2p(mut self) -> Self {
330 self.p2p = true;
331
332 self
333 }
334
335 /// The to-be-created connection will be a server using the given GUID.
336 ///
337 /// The to-be-created connection will wait for incoming client authentication handshake and
338 /// negotiation messages, for peer-to-peer communications after successful creation.
339 ///
340 /// This method is only available when the `p2p` feature is enabled.
341 ///
342 /// **NOTE:** This method is redundant when using [`Builder::authenticated_socket`] since the
343 /// latter already sets the GUID for the connection and zbus doesn't differentiate between a
344 /// server and a client connection, except for authentication.
345 #[cfg(feature = "p2p")]
346 pub fn server<G>(mut self, guid: G) -> Result<Self>
347 where
348 G: TryInto<Guid<'a>>,
349 G::Error: Into<Error>,
350 {
351 self.guid = Some(guid.try_into().map_err(Into::into)?);
352
353 Ok(self)
354 }
355
356 /// Set the capacity of the main (unfiltered) queue.
357 ///
358 /// Since typically you'd want to set this at instantiation time, you can set it through the
359 /// builder.
360 ///
361 /// # Example
362 ///
363 /// ```
364 /// # use std::error::Error;
365 /// # use zbus::connection::Builder;
366 /// # use zbus::block_on;
367 /// #
368 /// # block_on(async {
369 /// let conn = Builder::session()?
370 /// .max_queued(30)
371 /// .build()
372 /// .await?;
373 /// assert_eq!(conn.max_queued(), 30);
374 ///
375 /// # Ok::<(), zbus::Error>(())
376 /// # }).unwrap();
377 /// #
378 /// // Do something useful with `conn`..
379 /// # Ok::<_, Box<dyn Error + Send + Sync>>(())
380 /// ```
381 pub fn max_queued(mut self, max: usize) -> Self {
382 self.max_queued = Some(max);
383
384 self
385 }
386
387 /// Enable or disable the internal executor thread.
388 ///
389 /// The thread is enabled by default.
390 ///
391 /// See [Connection::executor] for more details.
392 pub fn internal_executor(mut self, enabled: bool) -> Self {
393 self.internal_executor = enabled;
394
395 self
396 }
397
398 /// Register a D-Bus [`Interface`] to be served at a given path.
399 ///
400 /// This is similar to [`zbus::ObjectServer::at`], except that it allows you to have your
401 /// interfaces available immediately after the connection is established. Typically, this is
402 /// exactly what you'd want. Also in contrast to [`zbus::ObjectServer::at`], this method will
403 /// replace any previously added interface with the same name at the same path.
404 ///
405 /// Standard interfaces (Peer, Introspectable, Properties) are added on your behalf. If you
406 /// attempt to add yours, [`Builder::build()`] will fail.
407 pub fn serve_at<P, I>(mut self, path: P, iface: I) -> Result<Self>
408 where
409 I: Interface,
410 P: TryInto<ObjectPath<'a>>,
411 P::Error: Into<Error>,
412 {
413 let path = path.try_into().map_err(Into::into)?;
414 let entry = self.interfaces.entry(path).or_default();
415 entry.insert(I::name(), ArcInterface::new(iface));
416 Ok(self)
417 }
418
419 /// Register a well-known name for this connection on the bus.
420 ///
421 /// This is similar to [`zbus::Connection::request_name`], except the name is requested as part
422 /// of the connection setup ([`Builder::build`]), immediately after interfaces
423 /// registered (through [`Builder::serve_at`]) are advertised. Typically this is
424 /// exactly what you want.
425 ///
426 /// The methods [`Builder::allow_name_replacements`] and [`Builder::replace_existing_names`]
427 /// allow to set the [`zbus::fdo::RequestNameFlags`] used to request the name.
428 pub fn name<W>(mut self, well_known_name: W) -> Result<Self>
429 where
430 W: TryInto<WellKnownName<'a>>,
431 W::Error: Into<Error>,
432 {
433 let well_known_name = well_known_name.try_into().map_err(Into::into)?;
434 self.names.insert(well_known_name);
435
436 Ok(self)
437 }
438
439 /// Whether the [`zbus::fdo::RequestNameFlags::AllowReplacement`] flag will be set when
440 /// requesting names.
441 pub fn allow_name_replacements(mut self, allow_replacement: bool) -> Self {
442 self.request_name_flags
443 .set(RequestNameFlags::AllowReplacement, allow_replacement);
444 self
445 }
446
447 /// Whether the [`zbus::fdo::RequestNameFlags::ReplaceExisting`] flag will be set when
448 /// requesting names.
449 pub fn replace_existing_names(mut self, replace_existing: bool) -> Self {
450 self.request_name_flags
451 .set(RequestNameFlags::ReplaceExisting, replace_existing);
452 self
453 }
454
455 /// Set the unique name of the connection.
456 ///
457 /// This is mainly provided for bus implementations. All other users should not need to use this
458 /// method. Hence why this method is only available when the `bus-impl` feature is enabled.
459 ///
460 /// # Panics
461 ///
462 /// It will panic if the connection is to a message bus as it's the bus that assigns
463 /// peers their unique names.
464 #[cfg(feature = "bus-impl")]
465 pub fn unique_name<U>(mut self, unique_name: U) -> Result<Self>
466 where
467 U: TryInto<crate::names::UniqueName<'a>>,
468 U::Error: Into<Error>,
469 {
470 if !self.p2p {
471 panic!("unique name can only be set for peer-to-peer connections");
472 }
473 let name = unique_name.try_into().map_err(Into::into)?;
474 self.unique_name = Some(name);
475
476 Ok(self)
477 }
478
479 /// Set a timeout for method calls.
480 ///
481 /// Method calls will return
482 /// `zbus::Error::InputOutput(std::io::Error(kind: ErrorKind::TimedOut))` if a client does not
483 /// receive an answer from a service in time.
484 pub fn method_timeout(mut self, timeout: std::time::Duration) -> Self {
485 self.method_timeout = Some(timeout);
486
487 self
488 }
489
490 /// Build the connection, consuming the builder.
491 ///
492 /// # Errors
493 ///
494 /// Until server-side bus connection is supported, attempting to build such a connection will
495 /// result in a [`Error::Unsupported`] error.
496 pub async fn build(self) -> Result<Connection> {
497 let (conn, _) = self.build_inner(false).await?;
498 Ok(conn)
499 }
500
501 /// Build the connection and return a [`MessageStream`] to receive messages from it.
502 ///
503 /// This is equivalent to [`Self::build`] followed by `MessageStream::from(&conn)`, except
504 /// that the stream is set up **before** the socket-reader task is started. No messages can
505 /// therefore be lost in the window between `build()` returning and `MessageStream::from`
506 /// being called. Use this when the peer may pipeline traffic right after authentication —
507 /// e.g. a bus implementation reading a `Hello` method call from a just-connected client.
508 ///
509 /// To get the [`Connection`] out of the returned stream, use `Connection::from(&stream)` —
510 /// this is cheap (an `Arc` clone).
511 ///
512 /// This method is only available when the `bus-impl` feature is enabled.
513 ///
514 /// # Example
515 ///
516 /// ```
517 /// # use futures_util::StreamExt;
518 /// # use zbus::{
519 /// # Connection, Guid, block_on,
520 /// # connection::{Builder, socket::Channel},
521 /// # message::Message,
522 /// # };
523 /// #
524 /// # block_on(async {
525 /// let guid = Guid::generate();
526 /// let (c1, c2) = Channel::pair();
527 ///
528 /// // Bus client sends a method call right away (simulates pipelining after auth).
529 /// let client = Builder::authenticated_socket(c1, guid.clone())
530 /// .unwrap()
531 /// .build()
532 /// .await
533 /// .unwrap();
534 /// let hello = Message::method_call("/org/freedesktop/DBus", "Hello")
535 /// .unwrap()
536 /// .destination("org.freedesktop.DBus")
537 /// .unwrap()
538 /// .build(&())
539 /// .unwrap();
540 /// client.send(&hello).await.unwrap();
541 ///
542 /// // Server builds *after* the client has already sent.
543 /// let mut stream = Builder::authenticated_socket(c2, guid)
544 /// .unwrap()
545 /// .p2p()
546 /// .build_message_stream()
547 /// .await
548 /// .unwrap();
549 ///
550 /// let msg = stream.next().await.unwrap().unwrap();
551 /// assert_eq!(msg.header().member().unwrap().as_str(), "Hello");
552 ///
553 /// let _conn: Connection = (&stream).into();
554 /// # });
555 /// ```
556 #[cfg(feature = "bus-impl")]
557 pub async fn build_message_stream(self) -> Result<MessageStream> {
558 let (conn, msg_receiver) = self.build_inner(true).await?;
559 let msg_receiver = msg_receiver.expect("build_inner(true) always returns Some");
560
561 Ok(MessageStream::for_subscription_channel(
562 msg_receiver,
563 None,
564 &conn,
565 ))
566 }
567
568 async fn build_inner(
569 self,
570 activate_msg_stream: bool,
571 ) -> Result<(Connection, Option<ActiveReceiver<Result<Message>>>)> {
572 let executor = Executor::new();
573 #[cfg(feature = "async-io")]
574 let internal_executor = self.internal_executor;
575 // Box the future as it's large and can cause stack overflow.
576 let conn =
577 Box::pin(executor.run(self.build_(executor.clone(), activate_msg_stream))).await?;
578
579 #[cfg(feature = "async-io")]
580 start_internal_executor(&executor, internal_executor)?;
581
582 Ok(conn)
583 }
584
585 async fn build_(
586 mut self,
587 executor: Executor<'static>,
588 activate_msg_stream: bool,
589 ) -> Result<(Connection, Option<ActiveReceiver<Result<Message>>>)> {
590 #[cfg(feature = "p2p")]
591 let is_bus_conn = !self.p2p;
592 #[cfg(not(feature = "p2p"))]
593 let is_bus_conn = true;
594
595 let mut auth = self.connect(is_bus_conn).await?;
596
597 // SAFETY: `Authenticated` is always built with these fields set to `Some`.
598 let socket_read = auth.socket_read.take().unwrap();
599 let already_received_bytes = auth.already_received_bytes.drain(..).collect();
600 #[cfg(unix)]
601 let already_received_fds = auth.already_received_fds.drain(..).collect();
602
603 let mut conn = Connection::new(auth, is_bus_conn, executor, self.method_timeout).await?;
604 conn.set_max_queued(self.max_queued.unwrap_or(DEFAULT_MAX_QUEUED));
605
606 if !self.interfaces.is_empty() {
607 let object_server = conn.ensure_object_server(false);
608 for (path, interfaces) in self.interfaces {
609 for (name, iface) in interfaces {
610 let added = object_server
611 .add_arc_interface(path.clone(), name.clone(), iface.clone())
612 .await?;
613 if !added {
614 return Err(Error::InterfaceExists(name.clone(), path.to_owned()));
615 }
616 }
617 }
618
619 let started_event = Event::new();
620 let listener = started_event.listen();
621 conn.start_object_server(Some(started_event));
622
623 listener.await;
624 }
625
626 // Set up a message receiver before the socket-reader task is spawned so that the
627 // caller cannot miss early messages due to a race with the reader task.
628 let msg_receiver = activate_msg_stream.then(|| conn.inner.msg_receiver.activate_cloned());
629
630 // Start the socket reader task.
631 conn.init_socket_reader(
632 socket_read,
633 already_received_bytes,
634 #[cfg(unix)]
635 already_received_fds,
636 );
637
638 for name in self.names {
639 conn.request_name_with_flags(name, self.request_name_flags)
640 .await?;
641 }
642
643 Ok((conn, msg_receiver))
644 }
645
646 fn new(target: Target) -> Self {
647 Self {
648 target: Some(target),
649 #[cfg(feature = "p2p")]
650 p2p: false,
651 max_queued: None,
652 guid: None,
653 internal_executor: true,
654 interfaces: HashMap::new(),
655 names: HashSet::new(),
656 auth_mechanism: None,
657 #[cfg(feature = "bus-impl")]
658 unique_name: None,
659 request_name_flags: BitFlags::default(),
660 method_timeout: None,
661 user_id: None,
662 }
663 }
664
665 async fn connect(&mut self, is_bus_conn: bool) -> Result<Authenticated> {
666 #[cfg(not(feature = "bus-impl"))]
667 let unique_name = None;
668 #[cfg(feature = "bus-impl")]
669 let unique_name = self.unique_name.take().map(Into::into);
670
671 #[allow(unused_mut)]
672 let (mut stream, server_guid, authenticated) = self.target_connect().await?;
673 if authenticated {
674 let (socket_read, socket_write) = stream.take();
675 Ok(Authenticated {
676 #[cfg(unix)]
677 cap_unix_fd: socket_read.can_pass_unix_fd(),
678 socket_read: Some(socket_read),
679 socket_write,
680 // SAFETY: `server_guid` is provided as arg of `Builder::authenticated_socket`.
681 server_guid: server_guid.unwrap(),
682 already_received_bytes: vec![],
683 unique_name,
684 #[cfg(unix)]
685 already_received_fds: vec![],
686 })
687 } else {
688 #[cfg(feature = "p2p")]
689 match self.guid.take() {
690 None => {
691 // SASL Handshake
692 Authenticated::client(
693 stream,
694 server_guid,
695 self.auth_mechanism,
696 is_bus_conn,
697 self.user_id,
698 )
699 .await
700 }
701 Some(guid) => {
702 if !self.p2p {
703 return Err(Error::Unsupported);
704 }
705
706 let creds = stream.read_mut().peer_credentials().await?;
707 #[cfg(unix)]
708 let client_uid = self.user_id.or_else(|| creds.unix_user_id());
709 #[cfg(windows)]
710 let client_sid = creds.into_windows_sid();
711
712 Authenticated::server(
713 stream,
714 guid.to_owned().into(),
715 #[cfg(unix)]
716 client_uid,
717 #[cfg(windows)]
718 client_sid,
719 self.auth_mechanism,
720 unique_name,
721 )
722 .await
723 }
724 }
725
726 #[cfg(not(feature = "p2p"))]
727 Authenticated::client(
728 stream,
729 server_guid,
730 self.auth_mechanism,
731 is_bus_conn,
732 self.user_id,
733 )
734 .await
735 }
736 }
737
738 async fn target_connect(&mut self) -> Result<(BoxedSplit, Option<OwnedGuid>, bool)> {
739 let mut authenticated = false;
740 let mut guid = None;
741 // SAFETY: `self.target` is always `Some` from the beginning and this method is only called
742 // once.
743 let split = match self.target.take().unwrap() {
744 #[cfg(all(unix, feature = "tokio"))]
745 Target::TokioUnixStream(stream) => stream.into(),
746 #[cfg(all(any(unix, windows), feature = "async-io"))]
747 Target::AsyncIoUnixStream(stream) => Async::new(stream)?.into(),
748 #[cfg(feature = "tokio")]
749 Target::TokioTcpStream(stream) => stream.into(),
750 #[cfg(feature = "async-io")]
751 Target::AsyncIoTcpStream(stream) => Async::new(stream)?.into(),
752 #[cfg(all(feature = "vsock", not(feature = "tokio-vsock")))]
753 Target::VsockStream(stream) => Async::new(stream)?.into(),
754 #[cfg(feature = "tokio-vsock")]
755 Target::VsockStream(stream) => stream.into(),
756 Target::Address(address) => {
757 guid = address.guid().map(|g| g.to_owned().into());
758 match address.connect().await? {
759 #[cfg(any(unix, feature = "async-io"))]
760 address::transport::Stream::Unix(split) => split,
761 #[cfg(unix)]
762 address::transport::Stream::Unixexec(split) => split,
763 address::transport::Stream::Tcp(split) => split,
764 #[cfg(any(feature = "vsock", feature = "tokio-vsock"))]
765 address::transport::Stream::Vsock(split) => split,
766 }
767 }
768 Target::Socket(stream) => stream,
769 Target::AuthenticatedSocket(stream) => {
770 authenticated = true;
771 guid = self.guid.take().map(Into::into);
772 stream
773 }
774 };
775
776 Ok((split, guid, authenticated))
777 }
778}
779
780/// Start the internal executor thread.
781///
782/// Returns a dummy task that keep the executor ticking thread from exiting due to absence of any
783/// tasks until socket reader task kicks in.
784#[cfg(feature = "async-io")]
785fn start_internal_executor(executor: &Executor<'static>, internal_executor: bool) -> Result<()> {
786 // tokio drives its own tasks; only the `async-io` backend needs this driver thread.
787 if internal_executor && executor.needs_internal_driver() {
788 let executor = executor.clone();
789 std::thread::Builder::new()
790 .name("zbus::Connection executor".into())
791 .spawn(move || {
792 crate::utils::block_on(async move {
793 // Run as long as there is a task to run.
794 while !executor.is_empty() {
795 executor.tick().await;
796 }
797 })
798 })?;
799 }
800
801 Ok(())
802}