Merge pull request #11 from hdoordt/dyn-delivery

Introduce DynDelivery to allow for combining buses with similar payloads

Author: hdoordt

src/chan/rpc/comm.rs

@@ -1,15 +1,15 @@
 use std::{any::type_name, marker::PhantomData};
 
-use futures::Stream;
+use futures::{Future, Stream, StreamExt};
 use lapin::BasicProperties;
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 use tracing::debug;
 use uuid::Uuid;
 
 use crate::{
-    error::Error, Bus, Channel, Consumer, Delivery, DirectBus, Publisher, ReplyError, Result,
-    RpcChannel,
+    error::Error, Bus, Channel, Consumer, Delivery, DirectBus, DynRpcDelivery, Publisher,
+    ReplyError, Result, RpcChannel,
 };
 
 use super::ReplyReceiver;
@@ -154,7 +154,7 @@ where
     /// Reply to a message that was sent by the receiver of the initial message.
     pub async fn reply_forth(
         &self,
-        back_payload: &B::ForthPayload,
+        forth_payload: &B::ForthPayload,
         chan: &impl Channel,
     ) -> Result<()> {
         let Some(correlation_uuid) = self.get_uuid() else {
@@ -166,15 +166,15 @@ where
 
         let correlation_uuid = correlation_uuid?;
 
-        let bytes = serde_json::to_vec(back_payload)?;
+        let bytes = serde_json::to_vec(forth_payload)?;
 
         debug!("Replying forth to message with correlation UUID {correlation_uuid}");
         chan.publish_with_properties(&bytes, reply_to, Default::default(), correlation_uuid)
             .await
     }
 
     /// Reply to a message that was sent by the receiver of the initial message and await
-    /// any further messages from the receiver.
+    /// multiple replies from the receiver.
     /// The messages can be obtained by calling [futures::StreamExt::next] on the returned [Stream].
     pub async fn reply_recv_many(
         &self,
@@ -210,6 +210,33 @@ where
             .await?;
         Ok(rx)
     }
+
+    /// Reply to a message that was sent by the receiver of the initial message and await
+    /// one reply from the receiver.
+    /// The message can be obtained by `await`ing the returned [Future].
+    pub async fn reply_recv(
+        &self,
+        back_payload: &B::BackPayload,
+        rpc_chan: &RpcChannel,
+    ) -> Result<impl Future<Output = Delivery<CommReply<B>>>> {
+        let rx = self.reply_recv_many(back_payload, rpc_chan).await?;
+        Ok(async move {
+            // As `rx` won't be dropped, we can assume that either
+            // this future never resolves, or it resolves to a `Some`
+            rx.take(1).next().await.unwrap()
+        })
+    }
+
+    /// Convert this [Delivery] into a [DynRpcDelivery]
+    pub fn into_dyn_comm(self) -> DynRpcDelivery<B::BackPayload, B::ForthPayload> {
+        DynRpcDelivery {
+            inner: crate::DynDelivery {
+                inner: self.inner,
+                _marker: PhantomData,
+            },
+            _reply: PhantomData,
+        }
+    }
 }
 
 #[cfg(test)]

src/chan/rpc/mod.rs

@@ -1,4 +1,11 @@
-use std::{any::type_name, fmt::Display, marker::PhantomData, sync::Arc, task::Poll};
+use std::{
+    any::type_name,
+    fmt::Display,
+    marker::PhantomData,
+    ops::{Deref, DerefMut},
+    sync::Arc,
+    task::Poll,
+};
 
 use async_trait::async_trait;
 use dashmap::DashMap;
@@ -10,10 +17,10 @@ use tokio::{
     sync::mpsc,
     task::{self, JoinHandle},
 };
-use tracing::debug;
+use tracing::{debug, warn};
 use uuid::Uuid;
 
-use crate::{delivery_uuid, error::Error, Bus, Connection, Delivery, Result};
+use crate::{delivery_uuid, error::Error, Bus, Connection, Delivery, DynDelivery, Result};
 
 use super::{direct::DirectBus, Channel, Consumer, Publisher};
 
@@ -104,11 +111,15 @@ impl RpcChannel {
                                         Err(_) => todo!("report error"),
                                     };
 
-                                    if let Some(tx) = pending_replies.get(&msg_id) {
-                                        tx.send(msg).unwrap();
-                                    } else {
-                                        todo!("Report absence of sender");
-                                    };
+                                    let forwarding_success =
+                                        if let Some(tx) = pending_replies.get(&msg_id) {
+                                            tx.send(msg).is_ok()
+                                        } else {
+                                            false
+                                        };
+                                    if forwarding_success {
+                                        warn!("Received reply cannot be forwarded due to dropped Receiver. UUID: {}", msg_id);
+                                    }
                                 }
                             });
                             // `forward_reply` should run to completion
@@ -247,7 +258,7 @@ where
         payload: &B::PublishPayload,
     ) -> Result<impl Future<Output = Option<Delivery<Reply<B>>>>> {
         let rx = self.publish_recv_many(args, payload).await?;
-        Ok(async { rx.take(1).next().await })
+        Ok(async move { rx.take(1).next().await })
     }
 }
 
@@ -276,6 +287,55 @@ where
     }
 }
 
+#[derive(Debug)]
+/// A Delivery that is not tied to a [Bus], but instead is generic over
+/// the publish and reply payload of the [RpcBus] or [RpcCommBus] associated with the [Delivery]
+/// it was converted from. It can be used to combine [Delivery]s that originate from different
+/// [RpcBus]es, that have identical `PublishPayload` and `ReplyPayload` types defined,
+/// and allow for deserializing the `PublishPayload` as well as replying with the `ReplyPayload`
+pub struct DynRpcDelivery<P, R> {
+    inner: DynDelivery<P>,
+    _reply: PhantomData<R>,
+}
+
+impl<'dp, 'dr, P, R> DynRpcDelivery<P, R>
+where
+    P: Deserialize<'dp> + Serialize,
+    R: Deserialize<'dr> + Serialize,
+{
+    /// Reply to a [DynDelivery]
+    pub async fn reply(&self, reply_payload: &R, chan: &impl Channel) -> Result<()> {
+        let Some(correlation_uuid) = self.inner.get_uuid() else {
+            return Err(Error::Reply(ReplyError::NoCorrelationUuid));
+        };
+        let Some(reply_to) = self.inner.inner.properties.reply_to().as_ref().map(|r | r.as_str()) else {
+            return Err(Error::Reply(ReplyError::NoReplyToConfigured))
+        };
+
+        let correlation_uuid = correlation_uuid?;
+
+        let bytes = serde_json::to_vec(reply_payload)?;
+
+        debug!("Replying forth to message with correlation UUID {correlation_uuid}");
+        chan.publish_with_properties(&bytes, reply_to, Default::default(), correlation_uuid)
+            .await
+    }
+}
+
+impl<P, R> Deref for DynRpcDelivery<P, R> {
+    type Target = DynDelivery<P>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl<P, R> DerefMut for DynRpcDelivery<P, R> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
 #[pin_project(PinnedDrop)]
 struct ReplyReceiver<B> {
     #[pin]

src/lib.rs

@@ -72,6 +72,14 @@ where
         }
         Ok(())
     }
+
+    /// Convert this [Delivery] into a [DynDelivery]
+    pub fn into_dyn(self) -> DynDelivery<B::PublishPayload> {
+        DynDelivery {
+            inner: self.inner,
+            _marker: PhantomData,
+        }
+    }
 }
 
 impl<B> From<lapin::message::Delivery> for Delivery<B> {
@@ -89,3 +97,29 @@ fn delivery_uuid(delivery: &lapin::message::Delivery) -> Option<Result<Uuid>> {
     };
     Some(Uuid::from_str(correlation_id.as_str()).map_err(Into::into))
 }
+
+#[derive(Debug)]
+/// A Delivery that is not tied to a [Bus], but instead is generic over
+/// the publish payload of the [Bus] associated with the [Delivery]
+/// it was converted from. It can be used to combine [Delivery]s that originate from different
+/// [Bus]es, that have identical `PublishPayload` types defined,
+/// and allow for deserializing the `PublishPayload`
+pub struct DynDelivery<P> {
+    inner: lapin::message::Delivery,
+    _marker: PhantomData<P>,
+}
+
+impl<'dp, P> DynDelivery<P>
+where
+    P: Deserialize<'dp> + Serialize,
+{
+    /// Get the message correlation [Uuid]
+    pub fn get_uuid(&self) -> Option<Result<Uuid>> {
+        delivery_uuid(&self.inner)
+    }
+
+    /// Deserialize and return the payload from the [DynDelivery]
+    pub fn get_payload(&'dp self) -> Result<P> {
+        Ok(serde_json::from_slice(&self.inner.data)?)
+    }
+}