Add receive_signal and receive_signal_parameters

Closes #1820

Author: swsnr

gio/src/dbus_connection.rs

@@ -6,7 +6,11 @@ use crate::{
     ffi, ActionGroup, DBusConnection, DBusInterfaceInfo, DBusMessage, DBusMethodInvocation,
     DBusSignalFlags, MenuModel,
 };
-use glib::{prelude::*, translate::*, WeakRef};
+use futures_channel::mpsc;
+use futures_core::{FusedStream, Stream};
+use futures_util::SinkExt;
+use glib::{prelude::*, translate::*, variant::VariantTypeMismatchError, WeakRef};
+use pin_project_lite::pin_project;
 
 pub trait DBusMethodCall: Sized {
     fn parse_call(
@@ -211,6 +215,77 @@ pub struct DBusSignalRef<'a> {
     pub parameters: &'a glib::Variant,
 }
 
+pin_project! {
+    // rustdoc-stripper-ignore-next
+    /// A subscribed stream.
+    ///
+    /// A stream which wraps an inner stream of type `S` while holding on to a
+    /// subscription handle `H` to keep a subscription alive.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct SubscribedSignalStream<H, S> {
+        #[pin]
+        stream: S,
+        subscription: H,
+    }
+}
+
+impl<S> SubscribedSignalStream<SignalSubscription, S> {
+    // rustdoc-stripper-ignore-next
+    /// Downgrade the inner signal subscription to a weak one.
+    ///
+    /// See [`SignalSubscription::downgrade`] and [`WeakSignalSubscription`].
+    pub fn downgrade(self) -> SubscribedSignalStream<WeakSignalSubscription, S> {
+        SubscribedSignalStream {
+            subscription: self.subscription.downgrade(),
+            stream: self.stream,
+        }
+    }
+}
+
+impl<S> SubscribedSignalStream<WeakSignalSubscription, S> {
+    // rustdoc-stripper-ignore-next
+    /// Upgrade the inner signal subscription to a strong one.
+    ///
+    /// See [`WeakSignalSubscription::upgrade`] and [`SignalSubscription`].
+    pub fn downgrade(self) -> Option<SubscribedSignalStream<SignalSubscription, S>> {
+        self.subscription
+            .upgrade()
+            .map(|subscription| SubscribedSignalStream {
+                subscription,
+                stream: self.stream,
+            })
+    }
+}
+
+impl<H, S> Stream for SubscribedSignalStream<H, S>
+where
+    S: Stream,
+{
+    type Item = S::Item;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let this = self.project();
+        this.stream.poll_next(cx)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.stream.size_hint()
+    }
+}
+
+impl<H, S> FusedStream for SubscribedSignalStream<H, S>
+where
+    S: FusedStream,
+{
+    fn is_terminated(&self) -> bool {
+        self.stream.is_terminated()
+    }
+}
+
 // rustdoc-stripper-ignore-next
 /// Build a registered DBus object, by handling different parts of DBus.
 #[must_use = "The builder must be built to be used"]
@@ -651,4 +726,90 @@ impl DBusConnection {
             );
         }
     }
+
+    // rustdoc-stripper-ignore-next
+    /// Subscribe to a D-Bus signal and receive signal emissions as a stream.
+    ///
+    /// See [`Self::signal_subscribe`] for arguments.  `map_signal` maps the
+    /// received signal to the stream's element.  If it returns `None` the
+    /// signal emission is not propagated downstream.
+    ///
+    /// The returned stream holds a strong reference to this D-Bus connection,
+    /// and unsubscribes from the signal when dropped.
+    ///
+    /// To avoid reference cycles you may wish to downgrade the returned
+    /// stream to hold only weak reference to the connection using
+    /// [`SubscribedSignalStream::downgrade`].
+    pub fn receive_signal<T: Send + 'static, F: Fn(DBusSignalRef) -> Option<T> + 'static>(
+        &self,
+        sender: Option<&str>,
+        interface_name: Option<&str>,
+        member: Option<&str>,
+        object_path: Option<&str>,
+        arg0: Option<&str>,
+        flags: DBusSignalFlags,
+        map_signal: F,
+    ) -> SubscribedSignalStream<SignalSubscription, impl Stream<Item = T> + use<T, F>> {
+        let (tx, rx) = mpsc::channel(1);
+        let subscription = self.subscribe_to_signal(
+            sender,
+            interface_name,
+            member,
+            object_path,
+            arg0,
+            flags,
+            move |signal| {
+                let mut tx = tx.clone();
+                if let Some(item) = map_signal(signal) {
+                    glib::spawn_future(async move {
+                        // We really don't care for the error here.
+                        let _: Result<(), mpsc::SendError> = tx.send(item).await;
+                    });
+                }
+            },
+        );
+        SubscribedSignalStream {
+            subscription,
+            stream: rx,
+        }
+    }
+
+    // rustdoc-stripper-ignore-next
+    /// Subscribe to a D-Bus signal and receive signal parameters as a stream.
+    ///
+    /// See [`Self::signal_subscribe`] for arguments.  Decode the parameters
+    /// as type `T`, and propagate either the decoded parameters, or the type
+    /// error if the signal parameters had a different type.
+    ///
+    /// The returned stream holds a strong reference to this D-Bus connection,
+    /// and unsubscribes from the signal when dropped.
+    ///
+    /// To avoid reference cycles you may wish to downgrade the returned
+    /// stream to hold only weak reference to the connection using
+    /// [`SubscribedSignalStream::downgrade`].
+    pub fn receive_signal_parameters<T>(
+        &self,
+        sender: Option<&str>,
+        interface_name: Option<&str>,
+        member: Option<&str>,
+        object_path: Option<&str>,
+        arg0: Option<&str>,
+        flags: DBusSignalFlags,
+    ) -> SubscribedSignalStream<
+        SignalSubscription,
+        impl Stream<Item = Result<T, VariantTypeMismatchError>> + use<T>,
+    >
+    where
+        T: FromVariant + Send + 'static,
+    {
+        self.receive_signal(
+            sender,
+            interface_name,
+            member,
+            object_path,
+            arg0,
+            flags,
+            |signal| Some(signal.parameters.try_get()),
+        )
+    }
 }

gio/src/lib.rs

@@ -33,7 +33,8 @@ pub use self::dbus::*;
 mod dbus_connection;
 pub use self::dbus_connection::{
     ActionGroupExportId, DBusSignalRef, FilterId, MenuModelExportId, RegistrationBuilder,
-    RegistrationId, SignalSubscription, SignalSubscriptionId, WatcherId, WeakSignalSubscription,
+    RegistrationId, SignalSubscription, SignalSubscriptionId, SubscribedSignalStream, WatcherId,
+    WeakSignalSubscription,
 };
 mod dbus_message;
 mod dbus_method_invocation;