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 futures_channel::mpsc;
+use futures_core::{FusedStream, Stream};
+use futures_util::SinkExt;
 use glib::{prelude::*, translate::*, WeakRef};
+use pin_project_lite::pin_project;
 
 pub trait DBusMethodCall: Sized {
     fn parse_call(
@@ -211,6 +215,112 @@ pub struct DBusSignalRef<'a> {
     pub parameters: &'a glib::Variant,
 }
 
+impl DBusSignalRef<'_> {
+    pub fn to_owned(&self) -> OwnedDBusSignal {
+        OwnedDBusSignal {
+            connection: self.connection.clone(),
+            sender_name: self.sender_name.to_owned(),
+            object_path: self.object_path.to_owned(),
+            interface_name: self.interface_name.to_owned(),
+            signal_name: self.signal_name.to_owned(),
+            parameters: self.parameters.clone(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct OwnedDBusSignal {
+    // rustdoc-stripper-ignore-next
+    /// The connection the signal was emitted on.
+    pub connection: DBusConnection,
+    // rustdoc-stripper-ignore-next
+    /// The bus name of the sender which emitted the signal.
+    pub sender_name: String,
+    // rustdoc-stripper-ignore-next
+    /// The path of the object on `sender` the signal was emitted from.
+    pub object_path: String,
+    // rustdoc-stripper-ignore-next
+    /// The interface the signal belongs to.
+    pub interface_name: String,
+    // rustdoc-stripper-ignore-next
+    /// The name of the emitted signal.
+    pub signal_name: String,
+    // rustdoc-stripper-ignore-next
+    /// Parameters the signal was emitted with.
+    pub parameters: 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 +761,91 @@ 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.
+    ///
+    /// 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(
+        &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 = OwnedDBusSignal> + use<>>
+    {
+        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();
+                let signal = signal.to_owned();
+                glib::spawn_future(async move {
+                    // We really don't care for the error here.
+                    let _: Result<(), mpsc::SendError> = tx.send(signal).await;
+                });
+            },
+        );
+        SubscribedSignalStream {
+            subscription,
+            stream: rx,
+        }
+    }
+
+    // rustdoc-stripper-ignore-next
+    /// Subscribe to a D-Bus signal and receive its parameters as a stream.
+    ///
+    /// See [`Self::signal_subscribe`] for arguments.
+    ///
+    /// 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(
+        &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 = glib::Variant> + use<>> {
+        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();
+                let parameters = signal.parameters.clone();
+                glib::spawn_future(async move {
+                    // We really don't care for the error here.
+                    let _: Result<(), mpsc::SendError> = tx.send(parameters).await;
+                });
+            },
+        );
+        SubscribedSignalStream {
+            subscription,
+            stream: rx,
+        }
+    }
 }

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;