refactor: Rename WithChannels, tx and rx

Cargo.toml

@@ -57,6 +57,7 @@ tokio = { workspace = true, features = ["full"] }
 thousands = "0.2.0"
 # macro tests
 trybuild = "1.0.104"
+testresult = "0.4.1"
 
 [features]
 # enable the remote transport

examples/compute.rs

@@ -7,11 +7,11 @@ use std::{
 use anyhow::bail;
 use futures_buffered::BufferedStreamExt;
 use irpc::{
-    channel::{oneshot, spsc},
+    channel::{mpsc, oneshot},
     rpc::{listen, Handler},
     rpc_requests,
     util::{make_client_endpoint, make_server_endpoint},
-    Client, LocalSender, Request, Service, WithChannels,
+    Client, LocalSender, Request, RequestSender, Service,
 };
 use n0_future::{
     stream::StreamExt,
@@ -59,13 +59,13 @@ enum ComputeRequest {
 #[rpc_requests(ComputeService, message = ComputeMessage)]
 #[derive(Serialize, Deserialize)]
 enum ComputeProtocol {
-    #[rpc(tx=oneshot::Sender<u128>)]
+    #[rpc(reply=oneshot::Sender<u128>)]
     Sqr(Sqr),
-    #[rpc(rx=spsc::Receiver<i64>, tx=oneshot::Sender<i64>)]
+    #[rpc(updates=mpsc::Receiver<i64>, reply=oneshot::Sender<i64>)]
     Sum(Sum),
-    #[rpc(tx=spsc::Sender<u64>)]
+    #[rpc(reply=mpsc::Sender<u64>)]
     Fibonacci(Fibonacci),
-    #[rpc(rx=spsc::Receiver<u64>, tx=spsc::Sender<u64>)]
+    #[rpc(updates=mpsc::Receiver<u64>, reply=mpsc::Sender<u64>)]
     Multiply(Multiply),
 }
 
@@ -76,10 +76,10 @@ struct ComputeActor {
 
 impl ComputeActor {
     pub fn local() -> ComputeApi {
-        let (tx, rx) = tokio::sync::mpsc::channel(128);
-        let actor = Self { recv: rx };
+        let (reply, request) = tokio::sync::mpsc::channel(128);
+        let actor = Self { recv: request };
         n0_future::task::spawn(actor.run());
-        let local = LocalSender::<ComputeMessage, ComputeService>::from(tx);
+        let local = LocalSender::<ComputeMessage, ComputeService>::from(reply);
         ComputeApi {
             inner: local.into(),
         }
@@ -99,34 +99,45 @@ impl ComputeActor {
         match msg {
             ComputeMessage::Sqr(sqr) => {
                 trace!("sqr {:?}", sqr);
-                let WithChannels {
-                    tx, inner, span, ..
+                let Request {
+                    reply,
+                    message,
+                    span,
+                    ..
                 } = sqr;
                 let _entered = span.enter();
-                let result = (inner.num as u128) * (inner.num as u128);
-                tx.send(result).await?;
+                let result = (message.num as u128) * (message.num as u128);
+                reply.send(result).await?;
             }
             ComputeMessage::Sum(sum) => {
                 trace!("sum {:?}", sum);
-                let WithChannels { rx, tx, span, .. } = sum;
+                let Request {
+                    updates,
+                    reply,
+                    span,
+                    ..
+                } = sum;
                 let _entered = span.enter();
-                let mut receiver = rx;
+                let mut receiver = updates;
                 let mut total = 0;
                 while let Some(num) = receiver.recv().await? {
                     total += num;
                 }
-                tx.send(total).await?;
+                reply.send(total).await?;
             }
             ComputeMessage::Fibonacci(fib) => {
                 trace!("fibonacci {:?}", fib);
-                let WithChannels {
-                    tx, inner, span, ..
+                let Request {
+                    reply,
+                    message,
+                    span,
+                    ..
                 } = fib;
                 let _entered = span.enter();
-                let mut sender = tx;
+                let sender = reply;
                 let mut a = 0u64;
                 let mut b = 1u64;
-                while a <= inner.max {
+                while a <= message.max {
                     sender.send(a).await?;
                     let next = a + b;
                     a = b;
@@ -135,17 +146,17 @@ impl ComputeActor {
             }
             ComputeMessage::Multiply(mult) => {
                 trace!("multiply {:?}", mult);
-                let WithChannels {
-                    rx,
-                    tx,
-                    inner,
+                let Request {
+                    updates,
+                    reply,
+                    message,
                     span,
                     ..
                 } = mult;
                 let _entered = span.enter();
-                let mut receiver = rx;
-                let mut sender = tx;
-                let multiplier = inner.initial;
+                let mut receiver = updates;
+                let sender = reply;
+                let multiplier = message.initial;
                 while let Some(num) = receiver.recv().await? {
                     sender.send(multiplier * num).await?;
                 }
@@ -171,13 +182,13 @@ impl ComputeApi {
         let Some(local) = self.inner.local() else {
             bail!("cannot listen on a remote service");
         };
-        let handler: Handler<ComputeProtocol> = Arc::new(move |msg, rx, tx| {
+        let handler: Handler<ComputeProtocol> = Arc::new(move |msg, updates, reply| {
             let local = local.clone();
             Box::pin(match msg {
-                ComputeProtocol::Sqr(msg) => local.send((msg, tx)),
-                ComputeProtocol::Sum(msg) => local.send((msg, tx, rx)),
-                ComputeProtocol::Fibonacci(msg) => local.send((msg, tx)),
-                ComputeProtocol::Multiply(msg) => local.send((msg, tx, rx)),
+                ComputeProtocol::Sqr(msg) => local.send((msg, reply)),
+                ComputeProtocol::Sum(msg) => local.send((msg, reply, updates)),
+                ComputeProtocol::Fibonacci(msg) => local.send((msg, reply)),
+                ComputeProtocol::Multiply(msg) => local.send((msg, reply, updates)),
             })
         });
         Ok(AbortOnDropHandle::new(task::spawn(listen(
@@ -188,44 +199,44 @@ impl ComputeApi {
     pub async fn sqr(&self, num: u64) -> anyhow::Result<oneshot::Receiver<u128>> {
         let msg = Sqr { num };
         match self.inner.request().await? {
-            Request::Local(request) => {
+            RequestSender::Local(sender) => {
                 let (tx, rx) = oneshot::channel();
-                request.send((msg, tx)).await?;
+                sender.send((msg, tx)).await?;
                 Ok(rx)
             }
-            Request::Remote(request) => {
-                let (_tx, rx) = request.write(msg).await?;
+            RequestSender::Remote(sender) => {
+                let (_tx, rx) = sender.write(msg).await?;
                 Ok(rx.into())
             }
         }
     }
 
-    pub async fn sum(&self) -> anyhow::Result<(spsc::Sender<i64>, oneshot::Receiver<i64>)> {
+    pub async fn sum(&self) -> anyhow::Result<(mpsc::Sender<i64>, oneshot::Receiver<i64>)> {
         let msg = Sum;
         match self.inner.request().await? {
-            Request::Local(request) => {
-                let (num_tx, num_rx) = spsc::channel(10);
+            RequestSender::Local(sender) => {
+                let (num_tx, num_rx) = mpsc::channel(10);
                 let (sum_tx, sum_rx) = oneshot::channel();
-                request.send((msg, sum_tx, num_rx)).await?;
+                sender.send((msg, sum_tx, num_rx)).await?;
                 Ok((num_tx, sum_rx))
             }
-            Request::Remote(request) => {
-                let (tx, rx) = request.write(msg).await?;
+            RequestSender::Remote(sender) => {
+                let (tx, rx) = sender.write(msg).await?;
                 Ok((tx.into(), rx.into()))
             }
         }
     }
 
-    pub async fn fibonacci(&self, max: u64) -> anyhow::Result<spsc::Receiver<u64>> {
+    pub async fn fibonacci(&self, max: u64) -> anyhow::Result<mpsc::Receiver<u64>> {
         let msg = Fibonacci { max };
         match self.inner.request().await? {
-            Request::Local(request) => {
-                let (tx, rx) = spsc::channel(128);
-                request.send((msg, tx)).await?;
+            RequestSender::Local(sender) => {
+                let (tx, rx) = mpsc::channel(128);
+                sender.send((msg, tx)).await?;
                 Ok(rx)
             }
-            Request::Remote(request) => {
-                let (_tx, rx) = request.write(msg).await?;
+            RequestSender::Remote(sender) => {
+                let (_tx, rx) = sender.write(msg).await?;
                 Ok(rx.into())
             }
         }
@@ -234,17 +245,17 @@ impl ComputeApi {
     pub async fn multiply(
         &self,
         initial: u64,
-    ) -> anyhow::Result<(spsc::Sender<u64>, spsc::Receiver<u64>)> {
+    ) -> anyhow::Result<(mpsc::Sender<u64>, mpsc::Receiver<u64>)> {
         let msg = Multiply { initial };
         match self.inner.request().await? {
-            Request::Local(request) => {
-                let (in_tx, in_rx) = spsc::channel(128);
-                let (out_tx, out_rx) = spsc::channel(128);
-                request.send((msg, out_tx, in_rx)).await?;
+            RequestSender::Local(sender) => {
+                let (in_tx, in_rx) = mpsc::channel(128);
+                let (out_tx, out_rx) = mpsc::channel(128);
+                sender.send((msg, out_tx, in_rx)).await?;
                 Ok((in_tx, out_rx))
             }
-            Request::Remote(request) => {
-                let (tx, rx) = request.write(msg).await?;
+            RequestSender::Remote(sender) => {
+                let (tx, rx) = sender.write(msg).await?;
                 Ok((tx.into(), rx.into()))
             }
         }
@@ -260,7 +271,7 @@ async fn local() -> anyhow::Result<()> {
     println!("Local: 5^2 = {}", rx.await?);
 
     // Test Sum
-    let (mut tx, rx) = api.sum().await?;
+    let (tx, rx) = api.sum().await?;
     tx.send(1).await?;
     tx.send(2).await?;
     tx.send(3).await?;
@@ -276,7 +287,7 @@ async fn local() -> anyhow::Result<()> {
     println!();
 
     // Test Multiply
-    let (mut in_tx, mut out_rx) = api.multiply(3).await?;
+    let (in_tx, mut out_rx) = api.multiply(3).await?;
     in_tx.send(2).await?;
     in_tx.send(4).await?;
     in_tx.send(6).await?;
@@ -311,7 +322,7 @@ async fn remote() -> anyhow::Result<()> {
     println!("Remote: 4^2 = {}", rx.await?);
 
     // Test Sum
-    let (mut tx, rx) = api.sum().await?;
+    let (tx, rx) = api.sum().await?;
     tx.send(4).await?;
     tx.send(5).await?;
     tx.send(6).await?;
@@ -327,7 +338,7 @@ async fn remote() -> anyhow::Result<()> {
     println!();
 
     // Test Multiply
-    let (mut in_tx, mut out_rx) = api.multiply(5).await?;
+    let (in_tx, mut out_rx) = api.multiply(5).await?;
     in_tx.send(1).await?;
     in_tx.send(2).await?;
     in_tx.send(3).await?;
@@ -380,7 +391,7 @@ async fn bench(api: ComputeApi, n: u64) -> anyhow::Result<()> {
     // Sequential streaming (using Multiply instead of MultiplyUpdate)
     {
         let t0 = std::time::Instant::now();
-        let (mut send, mut recv) = api.multiply(2).await?;
+        let (send, mut recv) = api.multiply(2).await?;
         let handle = tokio::task::spawn(async move {
             for i in 0..n {
                 send.send(i).await?;