Struct MessageStream

pub struct MessageStream { /* private fields */ }

A stream::Stream implementation that yields Message items.

You can convert a Connection to this type and back to Connection.

NOTE: You must ensure a MessageStream is continuously polled or you will experience hangs. If you don’t need to continuously poll the MessageStream but need to keep it around for later use, keep the connection around and convert it into a MessageStream when needed. The conversion is not an expensive operation so you don’t need to worry about performance, unless you do it very frequently. If you need to convert back and forth frequently, you may want to consider keeping both a connection and stream around.

Implementations

impl MessageStream

pub async fn for_match_rule<R>( rule: R, conn: &Connection, max_queued: Option<usize>, ) -> Result<Self>

where R: TryInto<OwnedMatchRule>, R::Error: Into<Error>,

Create a message stream for the given match rule.

If conn is a bus connection and match rule is for a signal, the match rule will be registered with the bus and queued for deregistration when the stream is dropped. If you’d like immediate deregistration, use AsyncDrop::async_drop. The reason match rules are only registered with the bus for signals is that D-Bus specification only allows signals to be broadcasted and unicast messages are always sent to their destination (regardless of any match rules registered by the destination) by the bus. Hence there is no need to register match rules for non-signal messages with the bus.

Having said that, stream created by this method can still very useful as it allows you to avoid needless task wakeups and simplify your stream consuming code.

You can optionally also request the capacity of the underlying message queue through max_queued. If specified, the capacity is guaranteed to be at least max_queued. If not specified, the default of 64 is assumed. The capacity can also be changed later through MessageStream::set_max_queued.

Example

use async_io::Timer;
use zbus::{AsyncDrop, Connection, MatchRule, MessageStream, fdo::NameOwnerChanged};
use futures_util::{TryStreamExt, future::select, future::Either::{Left, Right}, pin_mut};

let conn = Connection::session().await?;
let rule = MatchRule::builder()
    .msg_type(zbus::MessageType::Signal)
    .sender("org.freedesktop.DBus")?
    .interface("org.freedesktop.DBus")?
    .member("NameOwnerChanged")?
    .add_arg("org.freedesktop.zbus.MatchRuleStreamTest42")?
    .build();
let mut stream = MessageStream::for_match_rule(
    rule,
    &conn,
    // For such a specific match rule, we don't need a big queue.
    Some(1),
).await?;

let rule_str = "type='signal',sender='org.freedesktop.DBus',\
                interface='org.freedesktop.DBus',member='NameOwnerChanged',\
                arg0='org.freedesktop.zbus.MatchRuleStreamTest42'";
assert_eq!(
    stream.match_rule().map(|r| r.to_string()).as_deref(),
    Some(rule_str),
);

// We register 2 names, starting with the uninteresting one. If `stream` wasn't filtering
// messages based on the match rule, we'd receive method return call for each of these 2
// calls first.
//
// Note that the `NameOwnerChanged` signal will not be sent by the bus  for the first name
// we register since we setup an arg filter.
conn.request_name("org.freedesktop.zbus.MatchRuleStreamTest44")
    .await?;
conn.request_name("org.freedesktop.zbus.MatchRuleStreamTest42")
    .await?;

let msg = stream.try_next().await?.unwrap();
let signal = NameOwnerChanged::from_message(msg).unwrap();
assert_eq!(signal.args()?.name(), "org.freedesktop.zbus.MatchRuleStreamTest42");
stream.async_drop().await;

// Ensure the match rule is deregistered and this connection doesn't receive
// `NameOwnerChanged` signals.
let stream = MessageStream::from(&conn).try_filter_map(|msg| async move {
    Ok(NameOwnerChanged::from_message(msg))
});
conn.release_name("org.freedesktop.zbus.MatchRuleStreamTest42").await?;

pin_mut!(stream);
let next = stream.try_next();
pin_mut!(next);
let timeout = Timer::after(std::time::Duration::from_millis(50));
pin_mut!(timeout);
match select(next, timeout).await {
   Left((msg, _)) => unreachable!("unexpected message: {:?}", msg),
   Right((_, _)) => (),
}

Caveats

Since this method relies on MatchRule::matches, it inherits its caveats.

pub fn match_rule(&self) -> Option<MatchRule<'_>>

The associated match rule, if any.

pub fn max_queued(&self) -> usize

The maximum number of messages to queue for this stream.

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

Set maximum number of messages to queue for this stream.

After this call, the capacity is guaranteed to be at least max_queued.

Trait Implementations

impl AsyncDrop for MessageStream

fn async_drop<'async_trait>( self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait,

Perform the async cleanup.

impl Clone for MessageStream

fn clone(&self) -> MessageStream

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for MessageStream

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

Formats the value using the given formatter.

impl From<&Connection> for MessageStream

fn from(conn: &Connection) -> Self

Converts to this type from the input type.

impl From<&MessageStream> for Connection

fn from(stream: &MessageStream) -> Connection

Converts to this type from the input type.

impl From<Connection> for MessageStream

fn from(conn: Connection) -> Self

Converts to this type from the input type.

impl From<MessageStream> for Connection

fn from(stream: MessageStream) -> Connection

Converts to this type from the input type.

impl FusedStream for MessageStream

fn is_terminated(&self) -> bool

Returns true if the stream should no longer be polled.

impl OrderedStream for MessageStream

type Data = Result<Arc<Message>, Error>

The unordered data carried by this stream

type Ordering = MessageSequence

The type ordered by this stream.

fn poll_next_before( self: Pin<&mut Self>, cx: &mut Context<'_>, before: Option<&Self::Ordering>, ) -> Poll<PollResult<Self::Ordering, Self::Data>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if needed, and returning NoneBefore if it is known that the stream will not produce any more values ordered before the given point.

fn position_hint(&self) -> Option<MaybeBorrowed<'_, Self::Ordering>>

The minimum value of the ordering for any future items.

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

Returns the bounds on the remaining length of the stream.

impl Stream for MessageStream

type Item = Result<Arc<Message>, Error>

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.