Convenient API to create stream based on server-side filtering (match rules) #243

Closed zeenix closed 1 year ago

In GitLab by @MaxVerevkin on Jan 1, 2022, 17:20

use futures::StreamExt;
use tokio::task::JoinHandle;

type AnyResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync>>;

#[tokio::main]
async fn main() -> AnyResult<()> {
    let dbus1 = zbus::Connection::system().await?;
    let dbus2 = dbus1.clone();

    let task1: JoinHandle<AnyResult<()>> = tokio::spawn(async move {
        let proxy = zbus::fdo::DBusProxy::new(&dbus1).await?;
        proxy
            .add_match(
                "type='signal',\
                    path='/org/freedesktop/NetworkManager',\
                    interface='org.freedesktop.DBus.Properties',\
                    member='PropertiesChanged'",
            )
            .await?;
        let mut stream = zbus::MessageStream::from(dbus1);
        let mut cnt = 0;
        loop {
            if let Some(_) = stream.next().await {
                cnt += 1;
                eprintln!("task1 received {} messages", cnt);
            }
        }
    });

    let task2: JoinHandle<AnyResult<()>> = tokio::spawn(async move {
        let mut stream = zbus::MessageStream::from(dbus2);
        let mut cnt = 0;
        loop {
            if let Some(_) = stream.next().await {
                cnt += 1;
                eprintln!("task2 received {} messages", cnt);
            }
        }
    });

    let (a, b) = tokio::try_join!(task1, task2)?;
    a?;
    b?;
    Ok(())
}

This code opens a connection and gives each task it's own clone. The first task adds a match rule and then converts its connection into a message stream. The second task just converts its connection into a stream.

So what I would like to happen is: the first task receives it's matches, while second task doesn't. Is this possible without manual filtering of messages? I suppose I can just open two separate connections, but I wonder if I can get away with only once connection.

In GitLab by @MaxVerevkin on Jan 1, 2022, 19:04

Okay, after looking at SignalStream implementation I assume I do have to filter. I think it would be nice to have the ability to use SignalStream for custom matches.

SignalStream implements futures::stream::Stream so you can make use of futures::stream::StreamExt::filter to create the filtered signal stream you need. The signal API generated by dbus_proxy provides a getter to access the underlying message for any advanced filtering needs. Although I just noticed that the docs are missing and outdated on that. Will fix that next week before rolling out 2.0 stable.

In GitLab by @MaxVerevkin on Jan 2, 2022, 12:18

The issue is that I cannot create a SignalStream for a match without a sender or with a path_namespace instead of path. https://gitlab.freedesktop.org/dbus/zbus/-/issues/69 covers the creation of match rules, and I wonder if it can be extended to provide a filter / stream for created matches too? Something like:

let mut rule = MatchRule::new();
rule.msg_type = Some(MessagyType::Signal);
rule.path = Some("/path".try_into().unwrap());
rule.interface = Some("my.interface".try_into().unwrap());
let mut stream = dbus_conn.receive_matches(&[rule, other_rule]);

You only need to fidle with match rule if you want to filter on the the bus (broker), rather than client. In your case, it seems you need more client-side filtering, which is already possible.

In general, I agree that we should have a better match rule API. Typical use case is to receive signals given a destination, path and interface and that's why the high level (Proxy) API will only target that (unless enough people ask for a high-levle API for this or I'm convinced of this being a typical-enough use case).

@MaxVerevkin !589 addresses #69. I think it wouldn't be a lot of work to add this API on top of that work now and I wouldn't be against it.

mentioned in commit 714b4d20439a9475423cfbaacc542a0a1d04f6d4

mentioned in commit d7fab24557cb14f238d58a0d01aba1de341b009f

mentioned in commit c33228184454ae814678dcc4979bb32aea1bc88c

In GitLab by @MaxVerevkin on Jan 6, 2023, 18:53

@zeenix Thanks, the new match rule API looks nice, I started porting some of my projects to it :) However I noticed that impl ToString for MatchRule<'_> ignores arg0namespace which makes arg0namespace unusable. Will send a patch.

dbus2 / zbus-old

Convenient API to create stream based on server-side filtering (match rules) #243