💩 zb: Workaround for xdg-dbus-proxy's monotonic serial requirement

While [the underlying issue in xdg-dbus-proxy][1] has already been fixed,
it will take some time before the fix is released and is widely available.
So let's add a workaround for now that makes message creation and
sending an atomic operation, using an async semaphore if flatpak portal
is detected.

[1]: flatpak/xdg-dbus-proxy#46

zbus/src/connection/mod.rs

@@ -21,9 +21,10 @@ use futures_core::Future;
 use futures_util::StreamExt;
 
 use crate::{
-    async_lock::Mutex,
+    async_lock::{Mutex, Semaphore, SemaphorePermit},
     blocking,
     fdo::{self, ConnectionCredentials, RequestNameFlags, RequestNameReply},
+    is_flatpak,
     message::{Flags, Message, Type},
     proxy::CacheProperties,
     DBusError, Error, Executor, MatchRule, MessageStream, ObjectServer, OwnedGuid, OwnedMatchRule,
@@ -351,6 +352,8 @@ impl Connection {
         M::Error: Into<Error>,
         B: serde::ser::Serialize + zvariant::DynamicType,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut builder = Message::method(path, method_name)?;
         if let Some(sender) = self.unique_name() {
             builder = builder.sender(sender)?
@@ -404,6 +407,8 @@ impl Connection {
         M::Error: Into<Error>,
         B: serde::ser::Serialize + zvariant::DynamicType,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut b = Message::signal(path, interface, signal_name)?;
         if let Some(sender) = self.unique_name() {
             b = b.sender(sender)?;
@@ -424,6 +429,8 @@ impl Connection {
     where
         B: serde::ser::Serialize + zvariant::DynamicType,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut b = Message::method_reply(call)?;
         if let Some(sender) = self.unique_name() {
             b = b.sender(sender)?;
@@ -442,6 +449,8 @@ impl Connection {
         E: TryInto<ErrorName<'e>>,
         E::Error: Into<Error>,
     {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let mut b = Message::method_error(call, error_name)?;
         if let Some(sender) = self.unique_name() {
             b = b.sender(sender)?;
@@ -459,6 +468,8 @@ impl Connection {
         call: &zbus::message::Header<'_>,
         err: impl DBusError,
     ) -> Result<()> {
+        let _permit = acquire_serial_num_semaphore().await;
+
         let m = err.create_reply(call)?;
         self.send(&m).await
     }
@@ -1273,6 +1284,20 @@ enum NameStatus {
     Queued(#[allow(unused)] Task<()>),
 }
 
+static SERIAL_NUM_SEMAPHORE: Semaphore = Semaphore::new(1);
+
+// Make message creation and sending an atomic operation, using an async
+// semaphore if flatpak portal is detected to workaround an xdg-dbus-proxy issue:
+//
+// https://github.com/flatpak/xdg-dbus-proxy/issues/46
+async fn acquire_serial_num_semaphore() -> Option<SemaphorePermit<'static>> {
+    if is_flatpak() {
+        Some(SERIAL_NUM_SEMAPHORE.acquire().await)
+    } else {
+        None
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;