rename Msg to WithChannels again

Author: rklaehn

examples/compute.rs

@@ -15,7 +15,7 @@ use quic_rpc::{
     channel::{mpsc, oneshot},
     rpc::{listen, Handler, RemoteRead},
     util::{make_client_endpoint, make_server_endpoint},
-    LocalMpscChannel, Msg, Service, ServiceRequest, ServiceSender,
+    LocalMpscChannel, Service, ServiceRequest, ServiceSender, WithChannels,
 };
 use quic_rpc_derive::rpc_requests;
 use serde::{Deserialize, Serialize};
@@ -100,13 +100,13 @@ impl ComputeActor {
         match msg {
             ComputeMessage::Sqr(sqr) => {
                 trace!("sqr {:?}", sqr);
-                let Msg { tx, inner, .. } = sqr;
+                let WithChannels { tx, inner, .. } = sqr;
                 let result = (inner.num as u128) * (inner.num as u128);
                 tx.send(result).await?;
             }
             ComputeMessage::Sum(sum) => {
                 trace!("sum {:?}", sum);
-                let Msg { rx, tx, .. } = sum;
+                let WithChannels { rx, tx, .. } = sum;
                 let mut receiver = rx;
                 let mut total = 0;
                 while let Some(num) = receiver.recv().await? {
@@ -116,7 +116,7 @@ impl ComputeActor {
             }
             ComputeMessage::Fibonacci(fib) => {
                 trace!("fibonacci {:?}", fib);
-                let Msg { tx, inner, .. } = fib;
+                let WithChannels { tx, inner, .. } = fib;
                 let mut sender = tx;
                 let mut a = 0u64;
                 let mut b = 1u64;
@@ -129,7 +129,7 @@ impl ComputeActor {
             }
             ComputeMessage::Multiply(mult) => {
                 trace!("multiply {:?}", mult);
-                let Msg { rx, tx, inner } = mult;
+                let WithChannels { rx, tx, inner } = mult;
                 let mut receiver = rx;
                 let mut sender = tx;
                 let multiplier = inner.initial;
@@ -406,7 +406,6 @@ fn clear_line() -> io::Result<()> {
     Ok(())
 }
 
-
 // Simple benchmark sending oneshot senders via an mpsc channel
 pub async fn reference_bench(n: u64) -> anyhow::Result<()> {
     // Create an mpsc channel to send oneshot senders
@@ -437,10 +436,7 @@ pub async fn reference_bench(n: u64) -> anyhow::Result<()> {
         let rps = ((n as f64) / t0.elapsed().as_secs_f64()).round() as u64;
         assert_eq!(sum, 42 * n); // Each response is 42
         clear_line()?;
-        println!(
-            "Reference seq {} rps",
-            rps.separate_with_underscores()
-        );
+        println!("Reference seq {} rps", rps.separate_with_underscores());
     }
 
     // Parallel oneshot sends
@@ -456,10 +452,7 @@ pub async fn reference_bench(n: u64) -> anyhow::Result<()> {
         let rps = ((n as f64) / t0.elapsed().as_secs_f64()).round() as u64;
         assert_eq!(sum, 42 * n); // Each response is 42
         clear_line()?;
-        println!(
-            "Reference par {} rps",
-            rps.separate_with_underscores()
-        );
+        println!("Reference par {} rps", rps.separate_with_underscores());
     }
 
     Ok(())

examples/derive.rs

@@ -11,7 +11,7 @@ use quic_rpc::{
     channel::{mpsc, oneshot},
     rpc::{listen, Handler},
     util::{make_client_endpoint, make_server_endpoint},
-    LocalMpscChannel, Msg, Service, ServiceRequest, ServiceSender,
+    LocalMpscChannel, Service, ServiceRequest, ServiceSender, WithChannels,
 };
 // Import the macro
 use quic_rpc_derive::rpc_requests;
@@ -80,18 +80,18 @@ impl StorageActor {
         match msg {
             StorageMessage::Get(get) => {
                 info!("get {:?}", get);
-                let Msg { tx, inner, .. } = get;
+                let WithChannels { tx, inner, .. } = get;
                 tx.send(self.state.get(&inner.key).cloned()).await.ok();
             }
             StorageMessage::Set(set) => {
                 info!("set {:?}", set);
-                let Msg { tx, inner, .. } = set;
+                let WithChannels { tx, inner, .. } = set;
                 self.state.insert(inner.key, inner.value);
                 tx.send(()).await.ok();
             }
             StorageMessage::List(list) => {
                 info!("list {:?}", list);
-                let Msg { mut tx, .. } = list;
+                let WithChannels { mut tx, .. } = list;
                 for (key, value) in &self.state {
                     if tx.send(format!("{key}={value}")).await.is_err() {
                         break;

examples/storage.rs

@@ -10,7 +10,7 @@ use quic_rpc::{
     channel::{mpsc, none::NoReceiver, oneshot},
     rpc::{listen, Handler},
     util::{make_client_endpoint, make_server_endpoint},
-    Channels, LocalMpscChannel, Msg, Service, ServiceRequest, ServiceSender,
+    Channels, LocalMpscChannel, Service, ServiceRequest, ServiceSender, WithChannels,
 };
 use serde::{Deserialize, Serialize};
 use tracing::info;
@@ -59,9 +59,9 @@ enum StorageProtocol {
 
 #[derive(derive_more::From)]
 enum StorageMessage {
-    Get(Msg<Get, StorageService>),
-    Set(Msg<Set, StorageService>),
-    List(Msg<List, StorageService>),
+    Get(WithChannels<Get, StorageService>),
+    Set(WithChannels<Set, StorageService>),
+    List(WithChannels<List, StorageService>),
 }
 
 struct StorageActor {
@@ -93,18 +93,18 @@ impl StorageActor {
         match msg {
             StorageMessage::Get(get) => {
                 info!("get {:?}", get);
-                let Msg { tx, inner, .. } = get;
+                let WithChannels { tx, inner, .. } = get;
                 tx.send(self.state.get(&inner.key).cloned()).await.ok();
             }
             StorageMessage::Set(set) => {
                 info!("set {:?}", set);
-                let Msg { tx, inner, .. } = set;
+                let WithChannels { tx, inner, .. } = set;
                 self.state.insert(inner.key, inner.value);
                 tx.send(()).await.ok();
             }
             StorageMessage::List(list) => {
                 info!("list {:?}", list);
-                let Msg { mut tx, .. } = list;
+                let WithChannels { mut tx, .. } = list;
                 for (key, value) in &self.state {
                     if tx.send(format!("{key}={value}")).await.is_err() {
                         break;

quic-rpc-derive/src/lib.rs

@@ -63,8 +63,12 @@ pub fn rpc_requests(attr: TokenStream, item: TokenStream) -> TokenStream {
         _ => return error_tokens(input.span(), "RpcRequests can only be applied to enums"),
     };
 
+    // builder for the trait impls
     let mut additional_items = Vec::new();
+    // types to check for uniqueness
     let mut types = HashSet::new();
+    // variant names and types
+    let mut variants = Vec::new();
 
     for variant in &mut data_enum.variants {
         // Check field structure for every variant
@@ -77,6 +81,7 @@ pub fn rpc_requests(attr: TokenStream, item: TokenStream) -> TokenStream {
                 )
             }
         };
+        variants.push((variant.ident.clone(), request_type.clone()));
 
         if !types.insert(request_type.to_token_stream().to_string()) {
             return error_tokens(input_span, "Each variant must have a unique request type");
@@ -107,6 +112,7 @@ pub fn rpc_requests(attr: TokenStream, item: TokenStream) -> TokenStream {
             );
         }
 
+        // if there is no attr, the user has to impl Channels manually
         if let Some(attr) = rpc_attr {
             let args = match attr.parse_args::<NamedTypeArgs>() {
                 Ok(info) => info,
@@ -120,29 +126,11 @@ pub fn rpc_requests(attr: TokenStream, item: TokenStream) -> TokenStream {
         }
     }
 
-    let message_variants = data_enum
-        .variants
-        .iter()
-        .map(|variant| {
-            let variant_name = &variant.ident;
-
-            // Extract the inner type using let-else patterns
-            let Fields::Unnamed(fields) = &variant.fields else {
-                unreachable!()
-            };
-            let Some(field) = fields.unnamed.first() else {
-                unreachable!()
-            };
-            let Type::Path(type_path) = &field.ty else {
-                unreachable!()
-            };
-            let Some(last_segment) = type_path.path.segments.last() else {
-                unreachable!()
-            };
-            let inner_type = &last_segment.ident;
-
+    let message_variants = variants
+        .into_iter()
+        .map(|(variant_name, inner_type)| {
             quote! {
-                #variant_name(::quic_rpc::Msg<#inner_type, #service_name>)
+                #variant_name(::quic_rpc::WithChannels<#inner_type, #service_name>)
             }
         })
         .collect::<Vec<_>>();

src/lib.rs

@@ -1,6 +1,7 @@
 use std::{fmt::Debug, io, marker::PhantomData, ops::Deref};
 
 use channel::none::NoReceiver;
+use sealed::Sealed;
 use serde::{de::DeserializeOwned, Serialize};
 #[cfg(feature = "test")]
 pub mod util;
@@ -22,13 +23,17 @@ impl<T> RpcMessage for T where
 }
 
 /// Marker trait for a service
-pub trait Service: Debug + Clone {}
+pub trait Service: Send + Sync + Debug + Clone + 'static {}
+
+mod sealed {
+    pub trait Sealed {}
+}
 
 /// Marker trait for a sender
-pub trait Sender: Debug {}
+pub trait Sender: Debug + Sealed {}
 
 /// Marker trait for a receiver
-pub trait Receiver: Debug {}
+pub trait Receiver: Debug + Sealed {}
 
 /// Channels to be used for a message and service
 pub trait Channels<S: Service> {
@@ -136,6 +141,7 @@ pub mod channel {
             }
         }
 
+        impl<T> crate::sealed::Sealed for Sender<T> {}
         impl<T> crate::Sender for Sender<T> {}
 
         pub enum Receiver<T> {
@@ -183,6 +189,7 @@ pub mod channel {
             }
         }
 
+        impl<T> crate::sealed::Sealed for Receiver<T> {}
         impl<T> crate::Receiver for Receiver<T> {}
     }
 
@@ -278,6 +285,7 @@ pub mod channel {
             }
         }
 
+        impl<T> crate::sealed::Sealed for Sender<T> {}
         impl<T> crate::Sender for Sender<T> {}
 
         pub enum Receiver<T> {
@@ -309,30 +317,34 @@ pub mod channel {
             }
         }
 
+        impl<T> crate::sealed::Sealed for Receiver<T> {}
         impl<T> crate::Receiver for Receiver<T> {}
     }
 
     /// No channels, used when no communication is needed
     pub mod none {
-        use crate::{Receiver, Sender};
+        use crate::{sealed::Sealed, Receiver, Sender};
 
         #[derive(Debug)]
         pub struct NoSender;
-
+        impl Sealed for NoSender {}
         impl Sender for NoSender {}
 
         #[derive(Debug)]
         pub struct NoReceiver;
 
+        impl Sealed for NoReceiver {}
         impl Receiver for NoReceiver {}
     }
 }
 
-/// A wrapper for a message with channels to send and receive it
+/// A wrapper for a message with channels to send and receive it.
+/// This expands the protocol message to a full message that includes the
+/// active and unserializable channels.
 ///
 /// rx and tx can be set to an appropriate channel kind.
 #[derive(Debug)]
-pub struct Msg<I: Channels<S>, S: Service> {
+pub struct WithChannels<I: Channels<S>, S: Service> {
     /// The inner message.
     pub inner: I,
     /// The return channel to send the response to. Can be set to [`crate::channel::none::NoSender`] if not needed.
@@ -344,7 +356,7 @@ pub struct Msg<I: Channels<S>, S: Service> {
 /// Tuple conversion from inner message and tx/rx channels to a WithChannels struct
 ///
 /// For the case where you want both tx and rx channels.
-impl<I: Channels<S>, S: Service, Tx, Rx> From<(I, Tx, Rx)> for Msg<I, S>
+impl<I: Channels<S>, S: Service, Tx, Rx> From<(I, Tx, Rx)> for WithChannels<I, S>
 where
     I: Channels<S>,
     <I as Channels<S>>::Tx: From<Tx>,
@@ -363,7 +375,7 @@ where
 /// Tuple conversion from inner message and tx channel to a WithChannels struct
 ///
 /// For the very common case where you just need a tx channel to send the response to.
-impl<I, S, Tx> From<(I, Tx)> for Msg<I, S>
+impl<I, S, Tx> From<(I, Tx)> for WithChannels<I, S>
 where
     I: Channels<S, Rx = NoReceiver>,
     S: Service,
@@ -380,7 +392,7 @@ where
 }
 
 /// Deref so you can access the inner fields directly
-impl<I: Channels<S>, S: Service> Deref for Msg<I, S> {
+impl<I: Channels<S>, S: Service> Deref for WithChannels<I, S> {
     type Target = I;
 
     fn deref(&self) -> &Self::Target {
@@ -686,10 +698,10 @@ impl<M, R, S> From<LocalMpscChannel<M, S>> for ServiceRequest<M, R, S> {
 }
 
 impl<M: Send, S: Service> LocalMpscChannel<M, S> {
-    pub fn send<T>(&self, value: impl Into<Msg<T, S>>) -> SendFut<M>
+    pub fn send<T>(&self, value: impl Into<WithChannels<T, S>>) -> SendFut<M>
     where
         T: Channels<S>,
-        M: From<Msg<T, S>>,
+        M: From<WithChannels<T, S>>,
     {
         let value: M = value.into().into();
         SendFut::new(self.0.clone(), value)