refactor: use RemoteService trait

Author: Frando

examples/compute.rs

@@ -7,7 +7,7 @@ use anyhow::bail;
 use futures_buffered::BufferedStreamExt;
 use irpc::{
     channel::{mpsc, oneshot},
-    rpc::{listen, MessageWithChannels},
+    rpc::{listen, RemoteService},
     rpc_requests,
     util::{make_client_endpoint, make_server_endpoint},
     Client, LocalSender, Request, WithChannels,
@@ -164,7 +164,7 @@ impl ComputeApi {
         let Some(local) = self.inner.local() else {
             bail!("cannot listen on a remote service");
         };
-        let handler = ComputeMessage::forwarding_handler(local);
+        let handler = ComputeProtocol::forwarding_handler(local);
         Ok(AbortOnDropHandle::new(task::spawn(listen(
             endpoint, handler,
         ))))

examples/derive.rs

@@ -6,7 +6,7 @@ use std::{
 use anyhow::{Context, Result};
 use irpc::{
     channel::{mpsc, oneshot},
-    rpc::MessageWithChannels,
+    rpc::RemoteService,
     rpc_requests,
     util::{make_client_endpoint, make_server_endpoint},
     Client, LocalSender, WithChannels,
@@ -130,7 +130,7 @@ impl StorageApi {
         let local = self.inner.local().context("cannot listen on remote API")?;
         let join_handle = task::spawn(irpc::rpc::listen(
             endpoint,
-            StorageMessage::forwarding_handler(local),
+            StorageProtocol::forwarding_handler(local),
         ));
         Ok(AbortOnDropHandle::new(join_handle))
     }

irpc-derive/src/lib.rs

@@ -134,16 +134,16 @@ fn generate_message_from_wire_impl(
         });
 
     quote! {
-        impl ::irpc::rpc::MessageWithChannels for #message_enum_name {
-            type WireMessage = #proto_enum_name;
+        impl ::irpc::rpc::RemoteService for #proto_enum_name {
             fn from_wire(
                 msg: Self::WireMessage,
                 rx: ::irpc::rpc::quinn::RecvStream,
-                tx: ::irpc::rpc::quinn::SendStream) -> Self {
-                    match msg {
-                        #(#variants),*
-                    }
+                tx: ::irpc::rpc::quinn::SendStream
+            ) -> Self::Message {
+                match msg {
+                    #(#variants),*
                 }
+            }
         }
     }
 }

irpc-iroh/examples/auth.rs

@@ -132,7 +132,7 @@ mod storage {
     impl ProtocolHandler for StorageServer {
         async fn accept(&self, conn: Connection) -> Result<(), AcceptError> {
             let mut authed = false;
-            while let Some(msg) = read_request(&conn).await? {
+            while let Some(msg) = read_request::<StorageProtocol>(&conn).await? {
                 match msg {
                     StorageMessage::Auth(msg) => {
                         let WithChannels { inner, tx, .. } = msg;

irpc-iroh/examples/derive.rs

@@ -62,7 +62,7 @@ mod storage {
     use iroh::{protocol::ProtocolHandler, Endpoint};
     use irpc::{
         channel::{mpsc, oneshot},
-        rpc::MessageWithChannels,
+        rpc::RemoteService,
         rpc_requests, Client, LocalSender, WithChannels,
     };
     // Import the macro
@@ -171,7 +171,9 @@ mod storage {
                 .inner
                 .local()
                 .context("can not listen on remote service")?;
-            Ok(IrohProtocol::new(StorageMessage::forwarding_handler(local)))
+            Ok(IrohProtocol::new(StorageProtocol::forwarding_handler(
+                local,
+            )))
         }
 
         pub async fn get(&self, key: String) -> irpc::Result<Option<String>> {

irpc-iroh/src/lib.rs

@@ -10,7 +10,7 @@ use iroh::{
 use irpc::{
     channel::RecvError,
     rpc::{
-        Handler, MessageWithChannels, RemoteConnection, ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED,
+        Handler, RemoteConnection, RemoteService, ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED,
         MAX_MESSAGE_SIZE,
     },
     util::AsyncReadVarintExt,
@@ -142,13 +142,13 @@ pub async fn handle_connection<R: DeserializeOwned + 'static>(
     }
 }
 
-pub async fn read_request<M: MessageWithChannels>(
+pub async fn read_request<S: RemoteService>(
     connection: &Connection,
-) -> std::io::Result<Option<M>> {
+) -> std::io::Result<Option<S::Message>> {
     Ok(
-        match read_request_raw::<M::WireMessage>(connection).await? {
+        match read_request_raw::<S::WireMessage>(connection).await? {
             None => None,
-            Some((msg, rx, tx)) => Some(M::from_wire(msg, rx, tx)),
+            Some((msg, rx, tx)) => Some(S::from_wire(msg, rx, tx)),
         },
     )
 }

src/lib.rs

@@ -115,7 +115,7 @@ impl<T> RpcMessage for T where
 /// message type, and provides some type safety when sending messages.
 pub trait Service: Send + Sync + Debug + 'static {
     type WireMessage: Serialize + DeserializeOwned + Send + 'static;
-    type Message: Send + 'static;
+    type Message: Send + Unpin + 'static;
 }
 
 mod sealed {
@@ -1605,17 +1605,14 @@ pub mod rpc {
             + 'static,
     >;
 
-    pub trait MessageWithChannels: Send + Unpin + 'static {
-        type WireMessage: DeserializeOwned + Send;
-
-        fn from_wire(msg: Self::WireMessage, rx: quinn::RecvStream, tx: quinn::SendStream) -> Self;
-
-        fn forwarding_handler<S: Service<Message = Self>>(
-            local_sender: LocalSender<S>,
-        ) -> Handler<Self::WireMessage>
-        where
-            Self: Sized,
-        {
+    pub trait RemoteService: Service + Sized {
+        /// Creates a message from a wire message and a pair of quic streams.
+        fn from_wire(
+            msg: Self::WireMessage,
+            rx: quinn::RecvStream,
+            tx: quinn::SendStream,
+        ) -> Self::Message;
+        fn forwarding_handler(local_sender: LocalSender<Self>) -> Handler<Self::WireMessage> {
             Arc::new(move |msg, rx, tx| {
                 let msg = Self::from_wire(msg, rx, tx);
                 Box::pin(local_sender.send_raw(msg))
@@ -1661,13 +1658,13 @@ pub mod rpc {
         }
     }
 
-    pub async fn read_request<M: MessageWithChannels>(
+    pub async fn read_request<S: RemoteService>(
         connection: &quinn::Connection,
-    ) -> std::io::Result<Option<M>> {
+    ) -> std::io::Result<Option<S::Message>> {
         Ok(
-            match read_request_raw::<M::WireMessage>(connection).await? {
+            match read_request_raw::<S::WireMessage>(connection).await? {
                 None => None,
-                Some((msg, rx, tx)) => Some(M::from_wire(msg, rx, tx)),
+                Some((msg, rx, tx)) => Some(S::from_wire(msg, rx, tx)),
             },
         )
     }