feat!: Handler becomes a struct

Cargo.toml

@@ -33,6 +33,8 @@ postcard = { workspace = true, features = ["alloc", "use-std"], optional = true
 noq = { workspace = true, optional = true }
 # used as a buffer for serialization when using rpc
 smallvec = { version = "1.14.0", features = ["write"], optional = true }
+# used for the Debug impl of rpc::Handler
+derive_more = { version = "2", features = ["debug"], optional = true }
 # used in the test utils to generate noq endpoints
 rustls = { version = "0.23.5", default-features = false, features = ["std"], optional = true }
 # used in the test utils to generate noq endpoints
@@ -89,7 +91,7 @@ tracing-opentelemetry = "0.33"
 
 [features]
 # enable the remote transport
-rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "tokio/io-util"]
+rpc = ["dep:noq", "dep:postcard", "dep:smallvec", "dep:tracing", "dep:derive_more", "tokio/io-util"]
 # enable the iroh transport
 iroh = ["rpc", "dep:iroh", "dep:portable-atomic", "dep:getrandom"]
 # add helpers to create noq endpoints, for tests and for rpc on localhost

examples/0rtt.rs

@@ -240,7 +240,6 @@ mod ping {
         Client, WithChannels,
         channel::oneshot,
         iroh::{Iroh0RttProtocol, IrohProtocol},
-        rpc::RemoteService,
         rpc_requests,
     };
     use serde::{Deserialize, Serialize};
@@ -270,15 +269,15 @@ mod ping {
                 .inner
                 .as_local()
                 .context("can not listen on remote service")?;
-            Ok(Iroh0RttProtocol::new(EchoProtocol::remote_handler(local)))
+            Ok(Iroh0RttProtocol::new(local))
         }
 
         pub fn expose(self) -> Result<IrohProtocol<EchoProtocol>> {
             let local = self
                 .inner
                 .as_local()
                 .context("can not listen on remote service")?;
-            Ok(IrohProtocol::new(EchoProtocol::remote_handler(local)))
+            Ok(IrohProtocol::new(local))
         }
 
         pub async fn connect(

examples/compute.rs

@@ -9,7 +9,7 @@ use irpc::{
     Client, Request, WithChannels,
     channel::{mpsc, oneshot},
     noq::{listen, make_client_endpoint, make_server_endpoint},
-    rpc::RemoteService,
+    rpc::Handler,
     rpc_requests,
 };
 use n0_future::{
@@ -154,7 +154,7 @@ impl ComputeApi {
         let Some(local) = self.inner.as_local() else {
             bail!("cannot listen on a remote service");
         };
-        let handler = ComputeProtocol::remote_handler(local);
+        let handler = Handler::from_sender(local);
         Ok(AbortOnDropHandle::new(task::spawn(listen(
             endpoint, handler,
         ))))

examples/derive.rs

@@ -8,7 +8,6 @@ use irpc::{
     Client, WithChannels,
     channel::{mpsc, oneshot},
     noq::{make_client_endpoint, make_server_endpoint},
-    rpc::RemoteService,
     rpc_requests,
 };
 // Import the macro
@@ -130,10 +129,7 @@ impl StorageApi {
             .inner
             .as_local()
             .context("cannot listen on remote API")?;
-        let join_handle = task::spawn(irpc::noq::listen(
-            endpoint,
-            StorageProtocol::remote_handler(local),
-        ));
+        let join_handle = task::spawn(irpc::noq::listen(endpoint, local));
         Ok(AbortOnDropHandle::new(join_handle))
     }
 

examples/remote-and-local.rs

@@ -77,7 +77,6 @@ mod storage {
     use irpc::{
         Client, WithChannels,
         channel::{mpsc, oneshot},
-        rpc::RemoteService,
         rpc_requests,
     };
     use n0_future::task::AbortOnDropHandle;
@@ -183,7 +182,7 @@ mod storage {
                 .client
                 .as_local()
                 .context("can not listen on remote service")?;
-            Ok(IrohProtocol::new(StorageProtocol::remote_handler(local)))
+            Ok(IrohProtocol::new(local))
         }
 
         pub async fn get(&self, key: String) -> irpc::Result<Option<String>> {

examples/server-actor.rs

@@ -41,7 +41,7 @@ mod proto {
         let client = Client::<FooProtocol>::local(tx);
 
         let endpoint = Endpoint::bind(presets::N0).await?;
-        let protocol = IrohProtocol::with_sender(client.as_local().unwrap());
+        let protocol = IrohProtocol::new(client.as_local().unwrap());
         let router = Router::builder(endpoint).accept(ALPN, protocol).spawn();
         println!("endpoint id: {}", router.endpoint().id());
 

examples/span_propagation.rs

@@ -24,12 +24,12 @@
 //! After opening a trace, click "Trace Logs" in the view selector at the top right
 //! to open a log view that assembles logs from both client and server.
 
-use std::{sync::Arc, time::Duration};
+use std::time::Duration;
 
 use anyhow::Result;
 use clap::{Parser, Subcommand};
 use iroh::{Endpoint, EndpointId, endpoint::presets, protocol::Router};
-use irpc::{WithChannels, channel::oneshot, iroh::IrohProtocol, rpc::RemoteService, rpc_requests};
+use irpc::{WithChannels, channel::oneshot, iroh::IrohProtocol, rpc::Handler, rpc_requests};
 use opentelemetry::KeyValue;
 use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_sdk::{Resource, trace::SdkTracerProvider};
@@ -148,38 +148,35 @@ async fn server(otlp_endpoint: &str) -> Result<()> {
     // on `rpc_requests` above.
     let endpoint = Endpoint::bind(presets::N0).await?;
     let provider = init_tracing("example-server", endpoint.id(), otlp_endpoint)?;
-    let protocol = IrohProtocol::<Proto>::new(Arc::new(|req, rx, tx| {
-        Box::pin(async move {
-            let msg: Message = <Proto as RemoteService>::with_remote_channels(req, rx, tx);
-            match msg {
-                Message::Get(msg) => {
-                    // `WithChannels` carries a `span` field that the derive
-                    // macro builds as `info_span!("Get")` with its parent set
-                    // from the propagated remote context, so the handler's
-                    // work shows up under the originating client trace.
-                    let WithChannels {
-                        inner, tx, span, ..
-                    } = msg;
-                    // Hand the response off to a separate task to show that
-                    // `.instrument(span)` keeps work hooked into the same
-                    // trace even after the handler future returns.
-                    tokio::spawn(
-                        async move {
-                            info!(?inner, "handling request");
-                            // Simulate async work.
-                            tokio::time::sleep(Duration::from_millis(rand::random_range(20..200)))
-                                .await;
-                            let res = inner.0.to_uppercase();
-                            info!(?res, "generated response");
-                            tx.send(res).await.ok();
-                            info!("response sent");
-                        }
-                        .instrument(span),
-                    );
-                }
+    let protocol = IrohProtocol::<Proto>::new(Handler::sequential(|msg: Message| async move {
+        match msg {
+            Message::Get(msg) => {
+                // `WithChannels` carries a `span` field that the derive
+                // macro builds as `info_span!("Get")` with its parent set
+                // from the propagated remote context, so the handler's
+                // work shows up under the originating client trace.
+                let WithChannels {
+                    inner, tx, span, ..
+                } = msg;
+                // Hand the response off to a separate task to show that
+                // `.instrument(span)` keeps work hooked into the same
+                // trace even after the handler future returns.
+                tokio::spawn(
+                    async move {
+                        info!(?inner, "handling request");
+                        // Simulate async work.
+                        tokio::time::sleep(Duration::from_millis(rand::random_range(20..200)))
+                            .await;
+                        let res = inner.0.to_uppercase();
+                        info!(?res, "generated response");
+                        tx.send(res).await.ok();
+                        info!("response sent");
+                    }
+                    .instrument(span),
+                );
             }
-            Ok(())
-        })
+        }
+        Ok(())
     }));
     let router = Router::builder(endpoint).accept(ALPN, protocol).spawn();
 

examples/storage.rs

@@ -8,7 +8,7 @@ use irpc::{
     Channels, Client, Request, Service, WithChannels,
     channel::{mpsc, none::NoReceiver, oneshot},
     noq::{listen, make_client_endpoint, make_server_endpoint},
-    rpc::RemoteService,
+    rpc::{Handler, RemoteService},
 };
 use n0_future::task::{self, AbortOnDropHandle};
 use serde::{Deserialize, Serialize};
@@ -135,7 +135,7 @@ impl StorageApi {
         let Some(local) = self.inner.as_local() else {
             bail!("cannot listen on a remote service");
         };
-        let handler = StorageProtocol::remote_handler(local);
+        let handler = Handler::from_sender(local);
         Ok(AbortOnDropHandle::new(task::spawn(listen(
             endpoint, handler,
         ))))