refactor: rename spsc module to mpsc

Frando · Frando · commit a049272645e3 · 2025-06-19T12:21:58.000+02:00
diff --git a/examples/compute.rs b/examples/compute.rs
@@ -7,7 +7,7 @@ use std::{
 use anyhow::bail;
 use futures_buffered::BufferedStreamExt;
 use irpc::{
-    channel::{oneshot, spsc},
+    channel::{oneshot, mpsc},
     rpc::{listen, Handler},
     rpc_requests,
     util::{make_client_endpoint, make_server_endpoint},
@@ -61,11 +61,11 @@ enum ComputeRequest {
 enum ComputeProtocol {
     #[rpc(tx=oneshot::Sender<u128>)]
     Sqr(Sqr),
-    #[rpc(rx=spsc::Receiver<i64>, tx=oneshot::Sender<i64>)]
+    #[rpc(rx=mpsc::Receiver<i64>, tx=oneshot::Sender<i64>)]
     Sum(Sum),
-    #[rpc(tx=spsc::Sender<u64>)]
+    #[rpc(tx=mpsc::Sender<u64>)]
     Fibonacci(Fibonacci),
-    #[rpc(rx=spsc::Receiver<u64>, tx=spsc::Sender<u64>)]
+    #[rpc(rx=mpsc::Receiver<u64>, tx=mpsc::Sender<u64>)]
     Multiply(Multiply),
 }
 
@@ -200,11 +200,11 @@ impl ComputeApi {
         }
     }
 
-    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);
+                let (num_tx, num_rx) = mpsc::channel(10);
                 let (sum_tx, sum_rx) = oneshot::channel();
                 request.send((msg, sum_tx, num_rx)).await?;
                 Ok((num_tx, sum_rx))
@@ -216,11 +216,11 @@ impl ComputeApi {
         }
     }
 
-    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);
+                let (tx, rx) = mpsc::channel(128);
                 request.send((msg, tx)).await?;
                 Ok(rx)
             }
@@ -234,12 +234,12 @@ 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);
+                let (in_tx, in_rx) = mpsc::channel(128);
+                let (out_tx, out_rx) = mpsc::channel(128);
                 request.send((msg, out_tx, in_rx)).await?;
                 Ok((in_tx, out_rx))
             }
diff --git a/examples/derive.rs b/examples/derive.rs
@@ -6,7 +6,7 @@ use std::{
 
 use anyhow::{Context, Result};
 use irpc::{
-    channel::{oneshot, spsc},
+    channel::{oneshot, mpsc},
     rpc::Handler,
     rpc_requests,
     util::{make_client_endpoint, make_server_endpoint},
@@ -55,9 +55,9 @@ enum StorageProtocol {
     Get(Get),
     #[rpc(tx=oneshot::Sender<()>)]
     Set(Set),
-    #[rpc(tx=oneshot::Sender<u64>, rx=spsc::Receiver<(String, String)>)]
+    #[rpc(tx=oneshot::Sender<u64>, rx=mpsc::Receiver<(String, String)>)]
     SetMany(SetMany),
-    #[rpc(tx=spsc::Sender<String>)]
+    #[rpc(tx=mpsc::Sender<String>)]
     List(List),
 }
 
@@ -152,7 +152,7 @@ impl StorageApi {
         self.inner.rpc(Get { key }).await
     }
 
-    pub async fn list(&self) -> irpc::Result<spsc::Receiver<String>> {
+    pub async fn list(&self) -> irpc::Result<mpsc::Receiver<String>> {
         self.inner.server_streaming(List, 16).await
     }
 
@@ -162,7 +162,7 @@ impl StorageApi {
 
     pub async fn set_many(
         &self,
-    ) -> irpc::Result<(spsc::Sender<(String, String)>, oneshot::Receiver<u64>)> {
+    ) -> irpc::Result<(mpsc::Sender<(String, String)>, oneshot::Receiver<u64>)> {
         self.inner.client_streaming(SetMany, 4).await
     }
 }
diff --git a/examples/storage.rs b/examples/storage.rs
@@ -6,7 +6,7 @@ use std::{
 
 use anyhow::bail;
 use irpc::{
-    channel::{none::NoReceiver, oneshot, spsc},
+    channel::{none::NoReceiver, oneshot, mpsc},
     rpc::{listen, Handler},
     util::{make_client_endpoint, make_server_endpoint},
     Channels, Client, LocalSender, Request, Service, WithChannels,
@@ -36,7 +36,7 @@ struct List;
 
 impl Channels<StorageService> for List {
     type Rx = NoReceiver;
-    type Tx = spsc::Sender<String>;
+    type Tx = mpsc::Sender<String>;
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -157,11 +157,11 @@ impl StorageApi {
         }
     }
 
-    pub async fn list(&self) -> anyhow::Result<spsc::Receiver<String>> {
+    pub async fn list(&self) -> anyhow::Result<mpsc::Receiver<String>> {
         let msg = List;
         match self.inner.request().await? {
             Request::Local(request) => {
-                let (tx, rx) = spsc::channel(10);
+                let (tx, rx) = mpsc::channel(10);
                 request.send((msg, tx)).await?;
                 Ok(rx)
             }
diff --git a/irpc-iroh/examples/auth.rs b/irpc-iroh/examples/auth.rs
@@ -72,7 +72,7 @@ mod storage {
         Endpoint,
     };
     use irpc::{
-        channel::{oneshot, spsc},
+        channel::{oneshot, mpsc},
         Client, Service, WithChannels,
     };
     // Import the macro
@@ -122,9 +122,9 @@ mod storage {
         Get(Get),
         #[rpc(tx=oneshot::Sender<()>)]
         Set(Set),
-        #[rpc(tx=oneshot::Sender<u64>, rx=spsc::Receiver<(String, String)>)]
+        #[rpc(tx=oneshot::Sender<u64>, rx=mpsc::Receiver<(String, String)>)]
         SetMany(SetMany),
-        #[rpc(tx=spsc::Sender<String>)]
+        #[rpc(tx=mpsc::Sender<String>)]
         List(List),
     }
 
@@ -265,7 +265,7 @@ mod storage {
             self.inner.rpc(Get { key }).await
         }
 
-        pub async fn list(&self) -> Result<spsc::Receiver<String>, irpc::Error> {
+        pub async fn list(&self) -> Result<mpsc::Receiver<String>, irpc::Error> {
             self.inner.server_streaming(List, 10).await
         }
 
diff --git a/irpc-iroh/examples/derive.rs b/irpc-iroh/examples/derive.rs
@@ -60,7 +60,7 @@ mod storage {
     use anyhow::{Context, Result};
     use iroh::{protocol::ProtocolHandler, Endpoint};
     use irpc::{
-        channel::{oneshot, spsc},
+        channel::{oneshot, mpsc},
         rpc::Handler,
         rpc_requests, Client, LocalSender, Service, WithChannels,
     };
@@ -97,7 +97,7 @@ mod storage {
         Get(Get),
         #[rpc(tx=oneshot::Sender<()>)]
         Set(Set),
-        #[rpc(tx=spsc::Sender<String>)]
+        #[rpc(tx=mpsc::Sender<String>)]
         List(List),
     }
 
@@ -190,7 +190,7 @@ mod storage {
             self.inner.rpc(Get { key }).await
         }
 
-        pub async fn list(&self) -> irpc::Result<spsc::Receiver<String>> {
+        pub async fn list(&self) -> irpc::Result<mpsc::Receiver<String>> {
             self.inner.server_streaming(List, 10).await
         }
 
diff --git a/src/lib.rs b/src/lib.rs
@@ -127,9 +127,9 @@ pub trait Receiver: Debug + Sealed {}
 
 /// Trait to specify channels for a message and service
 pub trait Channels<S: Service> {
-    /// The sender type, can be either spsc, oneshot or none
+    /// The sender type, can be either mpsc, oneshot or none
     type Tx: Sender;
-    /// The receiver type, can be either spsc, oneshot or none
+    /// The receiver type, can be either mpsc, oneshot or none
     ///
     /// For many services, the receiver is not needed, so it can be set to [`NoReceiver`].
     type Rx: Receiver;
@@ -315,14 +315,14 @@ pub mod channel {
 
     /// SPSC channel, similar to tokio's mpsc channel
     ///
-    /// For the rpc case, the send side can not be cloned, hence spsc instead of mpsc.
-    pub mod spsc {
+    /// For the rpc case, the send side can not be cloned, hence mpsc instead of mpsc.
+    pub mod mpsc {
         use std::{fmt::Debug, future::Future, io, pin::Pin, sync::Arc};
 
         use super::{RecvError, SendError};
         use crate::RpcMessage;
 
-        /// Create a local spsc sender and receiver pair, with the given buffer size.
+        /// Create a local mpsc sender and receiver pair, with the given buffer size.
         ///
         /// This is currently using a tokio channel pair internally.
         pub fn channel<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
@@ -582,7 +582,7 @@ pub mod channel {
         impl crate::Receiver for NoReceiver {}
     }
 
-    /// Error when sending a oneshot or spsc message. For local communication,
+    /// Error when sending a oneshot or mpsc message. For local communication,
     /// the only thing that can go wrong is that the receiver has been dropped.
     ///
     /// For rpc communication, there can be any number of errors, so this is a
@@ -608,7 +608,7 @@ pub mod channel {
         }
     }
 
-    /// Error when receiving a oneshot or spsc message. For local communication,
+    /// Error when receiving a oneshot or mpsc message. For local communication,
     /// the only thing that can go wrong is that the sender has been closed.
     ///
     /// For rpc communication, there can be any number of errors, so this is a
@@ -871,24 +871,24 @@ impl<M, R, S> Client<M, R, S> {
         }
     }
 
-    /// Performs a request for which the server returns a spsc receiver.
+    /// Performs a request for which the server returns a mpsc receiver.
     pub fn server_streaming<Req, Res>(
         &self,
         msg: Req,
         local_response_cap: usize,
-    ) -> impl Future<Output = Result<channel::spsc::Receiver<Res>>> + Send + 'static
+    ) -> impl Future<Output = Result<channel::mpsc::Receiver<Res>>> + Send + 'static
     where
         S: Service,
         M: From<WithChannels<Req, S>> + Send + Sync + Unpin + 'static,
         R: From<Req> + Serialize + Send + Sync + 'static,
-        Req: Channels<S, Tx = channel::spsc::Sender<Res>, Rx = NoReceiver> + Send + 'static,
+        Req: Channels<S, Tx = channel::mpsc::Sender<Res>, Rx = NoReceiver> + Send + 'static,
         Res: RpcMessage,
     {
         let request = self.request();
         async move {
-            let recv: channel::spsc::Receiver<Res> = match request.await? {
+            let recv: channel::mpsc::Receiver<Res> = match request.await? {
                 Request::Local(request) => {
-                    let (tx, rx) = channel::spsc::channel(local_response_cap);
+                    let (tx, rx) = channel::mpsc::channel(local_response_cap);
                     request.send((msg, tx)).await?;
                     rx
                 }
@@ -911,26 +911,26 @@ impl<M, R, S> Client<M, R, S> {
         local_update_cap: usize,
     ) -> impl Future<
         Output = Result<(
-            channel::spsc::Sender<Update>,
+            channel::mpsc::Sender<Update>,
             channel::oneshot::Receiver<Res>,
         )>,
     >
     where
         S: Service,
         M: From<WithChannels<Req, S>> + Send + Sync + Unpin + 'static,
         R: From<Req> + Serialize + 'static,
-        Req: Channels<S, Tx = channel::oneshot::Sender<Res>, Rx = channel::spsc::Receiver<Update>>,
+        Req: Channels<S, Tx = channel::oneshot::Sender<Res>, Rx = channel::mpsc::Receiver<Update>>,
         Update: RpcMessage,
         Res: RpcMessage,
     {
         let request = self.request();
         async move {
             let (update_tx, res_rx): (
-                channel::spsc::Sender<Update>,
+                channel::mpsc::Sender<Update>,
                 channel::oneshot::Receiver<Res>,
             ) = match request.await? {
                 Request::Local(request) => {
-                    let (req_tx, req_rx) = channel::spsc::channel(local_update_cap);
+                    let (req_tx, req_rx) = channel::mpsc::channel(local_update_cap);
                     let (res_tx, res_rx) = channel::oneshot::channel();
                     request.send((msg, res_tx, req_rx)).await?;
                     (req_tx, res_rx)
@@ -947,32 +947,32 @@ impl<M, R, S> Client<M, R, S> {
         }
     }
 
-    /// Performs a request for which the client can send updates, and the server returns a spsc receiver.
+    /// Performs a request for which the client can send updates, and the server returns a mpsc receiver.
     pub fn bidi_streaming<Req, Update, Res>(
         &self,
         msg: Req,
         local_update_cap: usize,
         local_response_cap: usize,
-    ) -> impl Future<Output = Result<(channel::spsc::Sender<Update>, channel::spsc::Receiver<Res>)>>
+    ) -> impl Future<Output = Result<(channel::mpsc::Sender<Update>, channel::mpsc::Receiver<Res>)>>
            + Send
            + 'static
     where
         S: Service,
         M: From<WithChannels<Req, S>> + Send + Sync + Unpin + 'static,
         R: From<Req> + Serialize + Send + 'static,
-        Req: Channels<S, Tx = channel::spsc::Sender<Res>, Rx = channel::spsc::Receiver<Update>>
+        Req: Channels<S, Tx = channel::mpsc::Sender<Res>, Rx = channel::mpsc::Receiver<Update>>
             + Send
             + 'static,
         Update: RpcMessage,
         Res: RpcMessage,
     {
         let request = self.request();
         async move {
-            let (update_tx, res_rx): (channel::spsc::Sender<Update>, channel::spsc::Receiver<Res>) =
+            let (update_tx, res_rx): (channel::mpsc::Sender<Update>, channel::mpsc::Receiver<Res>) =
                 match request.await? {
                     Request::Local(request) => {
-                        let (update_tx, update_rx) = channel::spsc::channel(local_update_cap);
-                        let (res_tx, res_rx) = channel::spsc::channel(local_response_cap);
+                        let (update_tx, update_rx) = channel::mpsc::channel(local_update_cap);
+                        let (res_tx, res_rx) = channel::mpsc::channel(local_response_cap);
                         request.send((msg, res_tx, update_rx)).await?;
                         (update_tx, res_rx)
                     }
@@ -1120,7 +1120,7 @@ pub mod rpc {
         channel::{
             none::NoSender,
             oneshot,
-            spsc::{self, DynReceiver, DynSender},
+            mpsc::{self, DynReceiver, DynSender},
             RecvError, SendError,
         },
         util::{now_or_never, AsyncReadVarintExt, WriteVarintExt},
@@ -1290,9 +1290,9 @@ pub mod rpc {
         }
     }
 
-    impl<T: RpcMessage> From<quinn::RecvStream> for spsc::Receiver<T> {
+    impl<T: RpcMessage> From<quinn::RecvStream> for mpsc::Receiver<T> {
         fn from(read: quinn::RecvStream) -> Self {
-            spsc::Receiver::Boxed(Box::new(QuinnReceiver {
+            mpsc::Receiver::Boxed(Box::new(QuinnReceiver {
                 recv: read,
                 _marker: PhantomData,
             }))
@@ -1320,9 +1320,9 @@ pub mod rpc {
         }
     }
 
-    impl<T: RpcMessage> From<quinn::SendStream> for spsc::Sender<T> {
+    impl<T: RpcMessage> From<quinn::SendStream> for mpsc::Sender<T> {
         fn from(write: quinn::SendStream) -> Self {
-            spsc::Sender::Boxed(Arc::new(QuinnSender(tokio::sync::Mutex::new(
+            mpsc::Sender::Boxed(Arc::new(QuinnSender(tokio::sync::Mutex::new(
                 QuinnSenderState::Open(QuinnSenderInner {
                     send: write,
                     buffer: SmallVec::new(),
diff --git a/tests/mpsc_sender.rs b/tests/mpsc_sender.rs
