Struct zbus::azync::Connection

pub struct Connection { /* fields omitted */ }

The asynchronous sibling of zbus::Connection.

Most of the API is very similar to zbus::Connection, except it’s asynchronous. However, there are a few differences:

Sending Messages

For sending messages you can either use Connection::send_message method or make use of the Sink implementation. For latter, you might find SinkExt API very useful. Keep in mind that Connection will not manage the serial numbers (cookies) on the messages for you when they are sent through the Sink implementation. You can manually assign unique serial numbers to them using the Connection::assign_serial_num method before sending them off, if needed. Having said that, the Sink is mainly useful for sending out signals, as they do not expect a reply, and serial numbers are not very useful for signals either for the same reason.

Since you do not need exclusive access to a zbus::Connection to send messages on the bus, Sink is also implemented on &Connection.

Receiving Messages

Unlike zbus::Connection, there is no direct async equivalent of zbus::Connection::receive_message method provided. This is because the futures crate already provides a nice rich API that makes use of the stream::Stream implementation.

Each Connection instance maintains its own queue of incoming messages (storing the last max_queued() messages), so you can filter the stream for messages you care about and discard the rest. To avoid having multiple receivers filtering the same queue, Stream is only available with an exclusive (mutable) reference to a Connection; clone the Connection to get a new queue to use the Stream.

Caveats

At the moment, a simultaneous flush request from multiple tasks/threads could potentially create a busy loop, thus wasting CPU time. This limitation may be removed in the future.

Examples

Get the session bus ID

use zbus::azync::Connection;

let mut connection = Connection::session().await?;

let reply = connection
    .call_method(
        Some("org.freedesktop.DBus"),
        "/org/freedesktop/DBus",
        Some("org.freedesktop.DBus"),
        "GetId",
        &(),
    )
    .await?;

let id: &str = reply.body()?;
println!("Unique ID of the bus: {}", id);

Monitoring all messages

Let’s eavesdrop on the session bus 😈 using the Monitor interface:

use futures_util::stream::TryStreamExt;
use zbus::azync::Connection;

let mut connection = Connection::session().await?;

connection
    .call_method(
        Some("org.freedesktop.DBus"),
        "/org/freedesktop/DBus",
        Some("org.freedesktop.DBus.Monitoring"),
        "BecomeMonitor",
        &(&[] as &[&str], 0u32),
    )
    .await?;

while let Some(msg) = connection.try_next().await? {
    println!("Got message: {}", msg);
}

This should print something like:

Got message: Signal NameAcquired from org.freedesktop.DBus
Got message: Signal NameLost from org.freedesktop.DBus
Got message: Method call GetConnectionUnixProcessID from :1.1324
Got message: Error org.freedesktop.DBus.Error.NameHasNoOwner:
             Could not get PID of name ':1.1332': no such name from org.freedesktop.DBus
Got message: Method call AddMatch from :1.918
Got message: Method return from org.freedesktop.DBus

Implementations

impl Connection

pub async fn send_message(&self, msg: Message) -> Result<u32>

Send msg to the peer.

Unlike our Sink implementation, this method sets a unique (to this connection) serial number on the message before sending it off, for you.

On successfully sending off msg, the assigned serial number is returned.

pub async fn call_method<'d, 'p, 'i, 'm, D, P, I, M, B>(
    &self,
    destination: Option<D>,
    path: P,
    interface: Option<I>,
    method_name: M,
    body: &B
) -> Result<Arc<Message>> where
    D: TryInto<BusName<'d>>,
    P: TryInto<ObjectPath<'p>>,
    I: TryInto<InterfaceName<'i>>,
    M: TryInto<MemberName<'m>>,
    D::Error: Into<Error>,
    P::Error: Into<Error>,
    I::Error: Into<Error>,
    M::Error: Into<Error>,
    B: Serialize + Type, 

Send a method call.

Create a method-call message, send it over the connection, then wait for the reply.

On successful reply, an Ok(Message) is returned. On error, an Err is returned. D-Bus error replies are returned as Error::MethodError.

pub async fn emit_signal<'d, 'p, 'i, 'm, D, P, I, M, B>(
    &self,
    destination: Option<D>,
    path: P,
    interface: I,
    signal_name: M,
    body: &B
) -> Result<()> where
    D: TryInto<BusName<'d>>,
    P: TryInto<ObjectPath<'p>>,
    I: TryInto<InterfaceName<'i>>,
    M: TryInto<MemberName<'m>>,
    D::Error: Into<Error>,
    P::Error: Into<Error>,
    I::Error: Into<Error>,
    M::Error: Into<Error>,
    B: Serialize + Type, 

Emit a signal.

Create a signal message, and send it over the connection.

pub async fn reply<B>(&self, call: &Message, body: &B) -> Result<u32> where
    B: Serialize + Type, 

Reply to a message.

Given an existing message (likely a method call), send a reply back to the caller with the given body.

Returns the message serial number.

pub async fn reply_error<'e, E, B>(
    &self,
    call: &Message,
    error_name: E,
    body: &B
) -> Result<u32> where
    B: Serialize + Type,
    E: TryInto<ErrorName<'e>>,
    E::Error: Into<Error>, 

Reply an error to a message.

Given an existing message (likely a method call), send an error reply back to the caller with the given error_name and body.

Returns the message serial number.

pub fn is_bus(&self) -> bool

Checks if self is a connection to a message bus.

This will return false for p2p connections.

pub fn assign_serial_num(&self, msg: &mut Message) -> Result<u32>

Assigns a serial number to msg that is unique to this connection.

This method can fail if msg is corrupt.

pub fn unique_name(&self) -> Option<&OwnedUniqueName>

The unique name as assigned by the message bus or None if not a message bus connection.

pub fn max_queued(&self) -> usize

Max number of messages to queue.

pub fn set_max_queued(&mut self, max: usize)

Set the max number of messages to queue.

pub fn server_guid(&self) -> &str

The server’s GUID.

pub fn executor(&self) -> &Executor<'static>

The underlying executor.

When internal-executor feature is disabled, zbus will not spawn thread internally to run the executor. You’re responsible to continuously tick the executor. Failure to do so will result in hangs.

Examples

Here is how one would typically run the zbus executor through tokio’s single-threaded scheduler:

use zbus::azync::Connection;
use tokio::runtime;

runtime::Builder::new_current_thread()
       .build()
       .unwrap()
       .block_on(async {
    let conn = Connection::session().await.unwrap();
    {
       let conn = conn.clone();
       tokio::task::spawn(async move {
           loop {
               conn.executor().tick().await;
           }
       });
    }

    // All your other async code goes here.
});

pub async fn as_raw_fd(&self) -> RawFd

Get the raw file descriptor of this connection.

pub async fn session() -> Result<Self>

Create a Connection to the session/user message bus.

pub async fn system() -> Result<Self>

Create a Connection to the system-wide message bus.

Trait Implementations

impl Clone for Connection

fn clone(&self) -> Connection

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Connection

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl From<Connection> for Connection

fn from(conn: Connection) -> Self

Performs the conversion.

impl From<Connection> for Connection

fn from(conn: Connection) -> Self

Performs the conversion.

impl Sink<Message> for Connection

type Error = Error

The type of value produced by the sink when an error occurs.

fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Attempts to prepare the Sink to receive a value.

fn start_send(self: Pin<&mut Self>, msg: Message) -> Result<()>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Flush any remaining output from this sink.

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Flush any remaining output and close this sink, if necessary.

impl<'a> Sink<Message> for &'a Connection

type Error = Error

The type of value produced by the sink when an error occurs.

fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>>

Attempts to prepare the Sink to receive a value.

fn start_send(self: Pin<&mut Self>, msg: Message) -> Result<()>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Flush any remaining output from this sink.

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>>

Flush any remaining output and close this sink, if necessary.

impl Stream for Connection

type Item = Result<Arc<Message>>

Values yielded by the stream.

fn poll_next(
    self: Pin<&mut Self>,
    cx: &mut Context<'_>
) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.