Map transports not services

Cargo.toml

@@ -49,12 +49,13 @@ tracing-subscriber = "0.3.16"
 tempfile = "3.5.0"
 proc-macro2 = "1.0.66"
 futures-buffered = "0.2.4"
+testresult = "0.4.1"
+nested_enum_utils = "0.1.0"
 
 [features]
 hyper-transport = ["dep:flume", "dep:hyper", "dep:bincode", "dep:bytes", "dep:tokio-serde", "dep:tokio-util"]
 quinn-transport = ["dep:flume", "dep:quinn", "dep:bincode", "dep:tokio-serde", "dep:tokio-util"]
 flume-transport = ["dep:flume"]
-combined-transport = []
 macros = []
 default = ["flume-transport"]
 

examples/errors.rs

@@ -55,8 +55,8 @@ impl Fs {
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let fs = Fs;
-    let (server, client) = quic_rpc::transport::flume::service_connection::<IoService>(1);
-    let client = RpcClient::new(client);
+    let (server, client) = quic_rpc::transport::flume::connection(1);
+    let client = RpcClient::<IoService, _>::new(client);
     let server = RpcServer::new(server);
     let handle = tokio::task::spawn(async move {
         for _ in 0..1 {

examples/macro.rs

@@ -105,7 +105,7 @@ create_store_dispatch!(Store, dispatch_store_request);
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let (server, client) = flume::service_connection::<StoreService>(1);
+    let (server, client) = flume::connection(1);
     let server_handle = tokio::task::spawn(async move {
         let target = Store;
         run_server_loop(StoreService, server, target, dispatch_store_request).await

examples/modularize.rs

@@ -10,7 +10,9 @@
 use anyhow::Result;
 use futures_lite::StreamExt;
 use futures_util::SinkExt;
-use quic_rpc::{transport::flume, RpcClient, RpcServer, ServiceConnection, ServiceEndpoint};
+use quic_rpc::{
+    client::BoxedServiceConnection, transport::flume, RpcClient, RpcServer, ServiceEndpoint,
+};
 use tracing::warn;
 
 use app::AppService;
@@ -19,14 +21,14 @@ use app::AppService;
 async fn main() -> Result<()> {
     // Spawn an inmemory connection.
     // Could use quic equally (all code in this example is generic over the transport)
-    let (server_conn, client_conn) = flume::service_connection::<AppService>(1);
+    let (server_conn, client_conn) = flume::connection(1);
 
     // spawn the server
     let handler = app::Handler::default();
     tokio::task::spawn(run_server(server_conn, handler));
 
     // run a client demo
-    client_demo(client_conn).await?;
+    client_demo(BoxedServiceConnection::<AppService>::new(client_conn)).await?;
 
     Ok(())
 }
@@ -50,8 +52,8 @@ async fn run_server<C: ServiceEndpoint<AppService>>(server_conn: C, handler: app
         }
     }
 }
-pub async fn client_demo<C: ServiceConnection<AppService>>(conn: C) -> Result<()> {
-    let rpc_client = RpcClient::new(conn);
+pub async fn client_demo(conn: BoxedServiceConnection<AppService>) -> Result<()> {
+    let rpc_client = RpcClient::<AppService>::new(conn);
     let client = app::Client::new(rpc_client.clone());
 
     // call a method from the top-level app client
@@ -99,15 +101,12 @@ mod app {
     //!
     //! It could also easily compose services from other crates or internal modules.
 
+    use super::iroh;
     use anyhow::Result;
     use derive_more::{From, TryInto};
-    use quic_rpc::{
-        message::RpcMsg, server::RpcChannel, RpcClient, Service, ServiceConnection, ServiceEndpoint,
-    };
+    use quic_rpc::{message::RpcMsg, server::RpcChannel, RpcClient, Service, ServiceEndpoint};
     use serde::{Deserialize, Serialize};
 
-    use super::iroh;
-
     #[derive(Debug, Serialize, Deserialize, From, TryInto)]
     pub enum Request {
         Iroh(iroh::Request),
@@ -156,10 +155,14 @@ mod app {
         pub async fn handle_rpc_request<E: ServiceEndpoint<AppService>>(
             self,
             req: Request,
-            chan: RpcChannel<AppService, E, AppService>,
+            chan: RpcChannel<AppService, E>,
         ) -> Result<()> {
             match req {
-                Request::Iroh(req) => self.iroh.handle_rpc_request(req, chan.map()).await?,
+                Request::Iroh(req) => {
+                    self.iroh
+                        .handle_rpc_request(req, chan.map().boxed())
+                        .await?
+                }
                 Request::AppVersion(req) => chan.rpc(req, self, Self::on_version).await?,
             };
             Ok(())
@@ -171,20 +174,16 @@ mod app {
     }
 
     #[derive(Debug, Clone)]
-    pub struct Client<S: Service, C: ServiceConnection<S>> {
-        pub iroh: iroh::Client<S, C>,
-        client: RpcClient<AppService, C, S>,
+    pub struct Client {
+        pub iroh: iroh::Client,
+        client: RpcClient<AppService>,
     }
 
-    impl<S, C> Client<S, C>
-    where
-        S: Service,
-        C: ServiceConnection<S>,
-    {
-        pub fn new(client: RpcClient<AppService, C, S>) -> Self {
+    impl Client {
+        pub fn new(client: RpcClient<AppService>) -> Self {
             Self {
-                iroh: iroh::Client::new(client.clone().map()),
-                client,
+                client: client.clone(),
+                iroh: iroh::Client::new(client.map().boxed()),
             }
         }
 
@@ -202,7 +201,7 @@ mod iroh {
 
     use anyhow::Result;
     use derive_more::{From, TryInto};
-    use quic_rpc::{server::RpcChannel, RpcClient, Service, ServiceConnection, ServiceEndpoint};
+    use quic_rpc::{server::RpcChannel, RpcClient, Service};
     use serde::{Deserialize, Serialize};
 
     use super::{calc, clock};
@@ -233,38 +232,38 @@ mod iroh {
     }
 
     impl Handler {
-        pub async fn handle_rpc_request<S, E>(
+        pub async fn handle_rpc_request(
             self,
             req: Request,
-            chan: RpcChannel<IrohService, E, S>,
-        ) -> Result<()>
-        where
-            S: Service,
-            E: ServiceEndpoint<S>,
-        {
+            chan: RpcChannel<IrohService>,
+        ) -> Result<()> {
             match req {
-                Request::Calc(req) => self.calc.handle_rpc_request(req, chan.map()).await?,
-                Request::Clock(req) => self.clock.handle_rpc_request(req, chan.map()).await?,
+                Request::Calc(req) => {
+                    self.calc
+                        .handle_rpc_request(req, chan.map().boxed())
+                        .await?
+                }
+                Request::Clock(req) => {
+                    self.clock
+                        .handle_rpc_request(req, chan.map().boxed())
+                        .await?
+                }
             }
             Ok(())
         }
     }
 
     #[derive(Debug, Clone)]
-    pub struct Client<S, C> {
-        pub calc: calc::Client<S, C>,
-        pub clock: clock::Client<S, C>,
+    pub struct Client {
+        pub calc: calc::Client,
+        pub clock: clock::Client,
     }
 
-    impl<S, C> Client<S, C>
-    where
-        S: Service,
-        C: ServiceConnection<S>,
-    {
-        pub fn new(client: RpcClient<IrohService, C, S>) -> Self {
+    impl Client {
+        pub fn new(client: RpcClient<IrohService>) -> Self {
             Self {
-                calc: calc::Client::new(client.clone().map()),
-                clock: clock::Client::new(client.clone().map()),
+                calc: calc::Client::new(client.clone().map().boxed()),
+                clock: clock::Client::new(client.clone().map().boxed()),
             }
         }
     }
@@ -280,7 +279,7 @@ mod calc {
     use quic_rpc::{
         message::{ClientStreaming, ClientStreamingMsg, Msg, RpcMsg},
         server::RpcChannel,
-        RpcClient, Service, ServiceConnection, ServiceEndpoint,
+        RpcClient, Service,
     };
     use serde::{Deserialize, Serialize};
     use std::fmt::Debug;
@@ -337,15 +336,11 @@ mod calc {
     pub struct Handler;
 
     impl Handler {
-        pub async fn handle_rpc_request<S, E>(
+        pub async fn handle_rpc_request(
             self,
             req: Request,
-            chan: RpcChannel<CalcService, E, S>,
-        ) -> Result<()>
-        where
-            S: Service,
-            E: ServiceEndpoint<S>,
-        {
+            chan: RpcChannel<CalcService>,
+        ) -> Result<()> {
             match req {
                 Request::Add(req) => chan.rpc(req, self, Self::on_add).await?,
                 Request::Sum(req) => chan.client_streaming(req, self, Self::on_sum).await?,
@@ -373,16 +368,12 @@ mod calc {
     }
 
     #[derive(Debug, Clone)]
-    pub struct Client<S, C> {
-        client: RpcClient<CalcService, C, S>,
+    pub struct Client {
+        client: RpcClient<CalcService>,
     }
 
-    impl<S, C> Client<S, C>
-    where
-        C: ServiceConnection<S>,
-        S: Service,
-    {
-        pub fn new(client: RpcClient<CalcService, C, S>) -> Self {
+    impl Client {
+        pub fn new(client: RpcClient<CalcService>) -> Self {
             Self { client }
         }
         pub async fn add(&self, a: i64, b: i64) -> anyhow::Result<i64> {
@@ -403,7 +394,7 @@ mod clock {
     use quic_rpc::{
         message::{Msg, ServerStreaming, ServerStreamingMsg},
         server::RpcChannel,
-        RpcClient, Service, ServiceConnection, ServiceEndpoint,
+        RpcClient, Service,
     };
     use serde::{Deserialize, Serialize};
     use std::{
@@ -475,15 +466,11 @@ mod clock {
             h
         }
 
-        pub async fn handle_rpc_request<S, E>(
+        pub async fn handle_rpc_request(
             self,
             req: Request,
-            chan: RpcChannel<ClockService, E, S>,
-        ) -> Result<()>
-        where
-            S: Service,
-            E: ServiceEndpoint<S>,
-        {
+            chan: RpcChannel<ClockService>,
+        ) -> Result<()> {
             match req {
                 Request::Tick(req) => chan.server_streaming(req, self, Self::on_tick).await?,
             }
@@ -517,16 +504,12 @@ mod clock {
     }
 
     #[derive(Debug, Clone)]
-    pub struct Client<S, C> {
-        client: RpcClient<ClockService, C, S>,
+    pub struct Client {
+        client: RpcClient<ClockService>,
     }
 
-    impl<S, C> Client<S, C>
-    where
-        C: ServiceConnection<S>,
-        S: Service,
-    {
-        pub fn new(client: RpcClient<ClockService, C, S>) -> Self {
+    impl Client {
+        pub fn new(client: RpcClient<ClockService>) -> Self {
             Self { client }
         }
         pub async fn tick(&self) -> Result<BoxStream<Result<usize>>> {

examples/store.rs

@@ -184,7 +184,7 @@ async fn main() -> anyhow::Result<()> {
         }
     }
 
-    let (server, client) = flume::service_connection::<StoreService>(1);
+    let (server, client) = flume::connection(1);
     let client = RpcClient::<StoreService, _>::new(client);
     let server = RpcServer::<StoreService, _>::new(server);
     let server_handle = tokio::task::spawn(server_future(server));
@@ -237,7 +237,7 @@ async fn _main_unsugared() -> anyhow::Result<()> {
         type Req = u64;
         type Res = String;
     }
-    let (server, client) = flume::service_connection::<Service>(1);
+    let (server, client) = flume::connection::<u64, String>(1);
     let to_string_service = tokio::spawn(async move {
         let (mut send, mut recv) = server.accept().await?;
         while let Some(item) = recv.next().await {