refactor: try out a version without a macro

Author: Frando

examples/stream.rs

@@ -1,48 +1,30 @@
 use std::{
     collections::BTreeMap,
-    future::{Future, IntoFuture},
     net::{Ipv4Addr, SocketAddr, SocketAddrV4},
 };
 
 use anyhow::{Context, Result};
-use futures_util::future::BoxFuture;
 use irpc::{
-    channel::{mpsc, oneshot},
+    channel::oneshot,
     rpc::RemoteService,
     rpc_requests,
-    util::{
-        make_client_endpoint, make_server_endpoint, IrpcReceiverFutExt, MpscSenderExt, StreamItem,
-    },
+    util::{make_client_endpoint, make_server_endpoint, MpscSenderExt, Progress, StreamSender},
     Client, WithChannels,
 };
 // Import the macro
 use n0_future::{
     task::{self, AbortOnDropHandle},
-    Stream, StreamExt,
+    StreamExt,
 };
 use serde::{Deserialize, Serialize};
 use tracing::info;
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, thiserror::Error)]
+#[error("{message}")]
 struct Error {
     message: String,
 }
 
-impl std::error::Error for Error {}
-
-impl std::fmt::Display for Error {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.message)
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize, StreamItem)]
-enum GetItem {
-    Item(String),
-    Error(Error),
-    Done,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 struct Set {
     key: String,
@@ -61,7 +43,7 @@ struct Get {
 enum StorageProtocol {
     #[rpc(tx=oneshot::Sender<()>)]
     Set(Set),
-    #[rpc(tx=mpsc::Sender<GetItem>)]
+    #[rpc(tx=StreamSender<String, Error>)]
     Get(Get),
 }
 
@@ -70,31 +52,6 @@ struct StorageActor {
     state: BTreeMap<String, String>,
 }
 
-struct GetProgress {
-    fut: BoxFuture<'static, irpc::Result<mpsc::Receiver<GetItem>>>,
-}
-
-impl GetProgress {
-    pub fn new(
-        fut: impl Future<Output = irpc::Result<mpsc::Receiver<GetItem>>> + Send + 'static,
-    ) -> Self {
-        Self { fut: Box::pin(fut) }
-    }
-
-    pub fn stream(self) -> impl Stream<Item = anyhow::Result<String>> {
-        self.fut.into_stream()
-    }
-}
-
-impl IntoFuture for GetProgress {
-    type Output = anyhow::Result<String>;
-    type IntoFuture = BoxFuture<'static, Self::Output>;
-
-    fn into_future(self) -> Self::IntoFuture {
-        Box::pin(self.fut.try_collect())
-    }
-}
-
 impl StorageActor {
     pub fn spawn() -> StorageApi {
         let (tx, rx) = tokio::sync::mpsc::channel(1);
@@ -164,8 +121,12 @@ impl StorageApi {
         Ok(AbortOnDropHandle::new(join_handle))
     }
 
-    pub fn get(&self, key: String) -> GetProgress {
-        GetProgress::new(self.inner.server_streaming(Get { key }, 16))
+    pub fn get(&self, key: String) -> Progress<String, Error> {
+        Progress::new(self.inner.server_streaming(Get { key }, 16))
+    }
+
+    pub fn get_vec(&self, key: String) -> Progress<String, Error, Vec<String>> {
+        Progress::new(self.inner.server_streaming(Get { key }, 16))
     }
 
     pub async fn set(&self, key: String, value: String) -> irpc::Result<()> {
@@ -174,9 +135,13 @@ impl StorageApi {
 }
 
 async fn client_demo(api: StorageApi) -> Result<()> {
-    api.set("hello".to_string(), "world".to_string()).await?;
+    api.set("hello".to_string(), "world and all".to_string())
+        .await?;
     let value = api.get("hello".to_string()).await?;
-    println!("get: hello = {value:?}");
+    println!("get (string): hello = {value:?}");
+
+    let value = api.get_vec("hello".to_string()).await?;
+    println!("get (vec): hello = {value:?}");
 
     api.set("loremipsum".to_string(), "dolor sit amet".to_string())
         .await?;

irpc-derive/src/lib.rs

@@ -611,6 +611,7 @@ fn vis_pub() -> syn::Visibility {
     })
 }
 
+// TODO(Frando): Remove if the generics approach works out fine?
 #[proc_macro_derive(StreamItem)]
 pub fn derive_irpc_stream_item(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);

src/util.rs

@@ -450,11 +450,112 @@ mod now_or_never {
 pub(crate) use now_or_never::now_or_never;
 
 mod stream_item {
-    use std::{future::Future, io};
+    use std::{
+        future::{Future, IntoFuture},
+        io,
+        marker::PhantomData,
+    };
 
+    use futures_util::future::BoxFuture;
     use n0_future::{stream, Stream, StreamExt};
+    use serde::{Deserialize, Serialize};
+
+    use crate::{
+        channel::{mpsc, RecvError, SendError},
+        RpcMessage,
+    };
+
+    pub type StreamSender<T, E> = mpsc::Sender<Item<T, E>>;
+    pub type StreamReceiver<T, E> = mpsc::Receiver<Item<T, E>>;
+
+    #[derive(thiserror::Error, Debug)]
+    pub enum StreamError<E: std::error::Error> {
+        #[error(transparent)]
+        Transport(#[from] crate::Error),
+        #[error(transparent)]
+        Remote(E),
+    }
+
+    impl<E: std::error::Error> From<crate::channel::RecvError> for StreamError<E> {
+        fn from(value: crate::channel::RecvError) -> Self {
+            Self::Transport(value.into())
+        }
+    }
+
+    pub type StreamResult<T, E> = std::result::Result<T, StreamError<E>>;
+
+    pub struct Progress<
+        T: RpcMessage,
+        E: RpcMessage + std::error::Error,
+        C: Extend<T> + Default = T,
+    > {
+        fut: BoxFuture<'static, crate::Result<mpsc::Receiver<Item<T, E>>>>,
+        _collection_type: PhantomData<C>,
+    }
+
+    impl<T, E, C> Progress<T, E, C>
+    where
+        T: RpcMessage,
+        E: RpcMessage + std::error::Error,
+        C: Extend<T> + Default + Send,
+    {
+        pub fn new(
+            fut: impl Future<Output = crate::Result<mpsc::Receiver<Item<T, E>>>> + Send + 'static,
+        ) -> Self {
+            Self {
+                fut: Box::pin(fut),
+                _collection_type: PhantomData,
+            }
+        }
+
+        pub fn stream(self) -> impl Stream<Item = StreamResult<T, E>> {
+            self.fut.into_stream()
+        }
+    }
+
+    impl<T, E, C> IntoFuture for Progress<T, E, C>
+    where
+        T: RpcMessage,
+        E: RpcMessage + std::error::Error,
+        C: Default + Extend<T> + Send + 'static,
+    {
+        type Output = StreamResult<C, E>;
+        type IntoFuture = BoxFuture<'static, Self::Output>;
 
-    use crate::channel::{mpsc, RecvError, SendError};
+        fn into_future(self) -> Self::IntoFuture {
+            Box::pin(self.fut.try_collect())
+        }
+    }
+
+    #[derive(Debug, Serialize, Deserialize, Clone)]
+    pub enum Item<T, E> {
+        Ok(T),
+        Err(E),
+        Done,
+    }
+
+    impl<T: RpcMessage, E: RpcMessage + std::error::Error> StreamItem for Item<T, E> {
+        type Item = T;
+        type Error = E;
+        fn into_result_opt(self) -> Option<Result<Self::Item, Self::Error>> {
+            match self {
+                Item::Ok(item) => Some(Ok(item)),
+                Item::Err(error) => Some(Err(error)),
+                Item::Done => None,
+            }
+        }
+
+        fn from_result(item: std::result::Result<Self::Item, Self::Error>) -> Self {
+            match item {
+                Ok(item) => Self::Ok(item),
+                Err(err) => Self::Err(err),
+            }
+        }
+
+        fn done() -> Self {
+            Self::Done
+        }
+    }
 
     /// Trait for an enum that has three variants, item, error, and done.
     ///
@@ -463,9 +564,9 @@ mod stream_item {
     /// for successful end of stream.
     pub trait StreamItem: crate::RpcMessage {
         /// The error case of the item enum.
-        type Error;
+        type Error: crate::RpcMessage + std::error::Error;
         /// The item case of the item enum.
-        type Item;
+        type Item: crate::RpcMessage;
         /// Converts the stream item into either None for end of stream, or a Result
         /// containing the item or an error. Error is assumed as a termination, so
         /// if you get error you won't get an additional end of stream marker.
@@ -481,7 +582,6 @@ mod stream_item {
         ///
         /// This will convert items and errors into the item enum type, and add
         /// a done marker if the stream ends without an error.
-        #[allow(dead_code)]
         fn forward_stream(
             self,
             stream: impl Stream<Item = std::result::Result<T::Item, T::Error>>,
@@ -538,9 +638,7 @@ mod stream_item {
         fn try_collect<C, E>(self) -> impl Future<Output = std::result::Result<C, E>>
         where
             C: Default + Extend<T::Item>,
-            E: From<T::Error>,
-            E: From<crate::Error>,
-            E: From<RecvError>;
+            E: From<StreamError<T::Error>>;
 
         /// Converts the receiver returned by this future into a stream of items,
         /// where each item is either a successful item or an error.
@@ -550,9 +648,7 @@ mod stream_item {
         /// first item and then terminate.
         fn into_stream<E>(self) -> impl Stream<Item = std::result::Result<T::Item, E>>
         where
-            E: From<T::Error>,
-            E: From<crate::Error>,
-            E: From<RecvError>;
+            E: From<StreamError<T::Error>>;
     }
 
     impl<T, F> IrpcReceiverFutExt<T> for F
@@ -563,9 +659,7 @@ mod stream_item {
         async fn try_collect<C, E>(self) -> std::result::Result<C, E>
         where
             C: Default + Extend<T::Item>,
-            E: From<T::Error>,
-            E: From<crate::Error>,
-            E: From<RecvError>,
+            E: From<StreamError<T::Error>>,
         {
             let mut items = C::default();
             let mut stream = self.into_stream::<E>();
@@ -580,9 +674,7 @@ mod stream_item {
 
         fn into_stream<E>(self) -> impl Stream<Item = std::result::Result<T::Item, E>>
         where
-            E: From<T::Error>,
-            E: From<crate::Error>,
-            E: From<RecvError>,
+            E: From<StreamError<T::Error>>,
         {
             enum State<S, T> {
                 Init(S),
@@ -597,24 +689,22 @@ mod stream_item {
             ) -> Option<(std::result::Result<T::Item, E>, State<S, T>)>
             where
                 T: StreamItem,
-                E: From<T::Error>,
-                E: From<crate::Error>,
-                E: From<RecvError>,
+                E: From<StreamError<T::Error>>,
             {
                 match rx.recv().await {
                     Ok(Some(item)) => match item.into_result_opt()? {
                         Ok(i) => Some((Ok(i), State::Receiving(rx))),
-                        Err(e) => Some((Err(E::from(e)), State::Done)),
+                        Err(e) => Some((Err(E::from(StreamError::Remote(e))), State::Done)),
                     },
-                    Ok(None) => Some((Err(E::from(eof())), State::Done)),
-                    Err(e) => Some((Err(E::from(e)), State::Done)),
+                    Ok(None) => Some((Err(E::from(StreamError::from(eof()))), State::Done)),
+                    Err(e) => Some((Err(E::from(StreamError::from(e))), State::Done)),
                 }
             }
             Box::pin(stream::unfold(State::Init(self), |state| async move {
                 match state {
                     State::Init(fut) => match fut.await {
                         Ok(rx) => process_recv(rx).await,
-                        Err(e) => Some((Err(E::from(e)), State::Done)),
+                        Err(e) => Some((Err(E::from(StreamError::from(e))), State::Done)),
                     },
                     State::Receiving(rx) => process_recv(rx).await,
                     State::Done => None,
@@ -629,4 +719,6 @@ mod stream_item {
 pub use irpc_derive::StreamItem;
 #[cfg(feature = "stream")]
 #[cfg_attr(quicrpc_docsrs, doc(cfg(feature = "stream")))]
-pub use stream_item::{IrpcReceiverFutExt, MpscSenderExt, StreamItem};
+pub use stream_item::{
+    IrpcReceiverFutExt, Item, MpscSenderExt, Progress, StreamItem, StreamReceiver, StreamSender,
+};