diff --git a/src/channel.rs b/src/channel.rs index 39170ef..be8d048 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -18,6 +18,9 @@ pub mod oneshot; pub enum SendError { /// The receiver has been closed. This is the only error that can occur /// for local communication. + /// + /// For remote communication, this means that the receiver stopped the + /// stream with code 0, for example because it was dropped. #[error("Receiver closed")] ReceiverClosed, /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]). diff --git a/src/rpc.rs b/src/rpc.rs index 95c5d40..e979b8f 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -211,6 +211,8 @@ impl From for SendError { { e!(SendError::MaxMessageSizeExceeded) } + // A receiver that is dropped stops the stream with code 0. + noq::WriteError::Stopped(code) if code == 0u32.into() => e!(SendError::ReceiverClosed), _ => e!(SendError::Io, io::Error::from(err)), } } diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs index 022a9a4..9974c34 100644 --- a/tests/mpsc_channel.rs +++ b/tests/mpsc_channel.rs @@ -6,15 +6,19 @@ use std::{ }; use irpc::{ + Client, WithChannels, channel::{ SendError, mpsc::{self, Receiver, RecvError}, + oneshot, }, - rpc::ERROR_CODE_DECODE_FAILED, + rpc::{ERROR_CODE_DECODE_FAILED, Handler}, + rpc_requests, util::AsyncWriteVarintExt, }; use n0_error::e; use noq::Endpoint; +use serde::{Deserialize, Serialize}; use testresult::TestResult; use tokio::time::timeout; @@ -120,6 +124,92 @@ async fn mpsc_sender_clone_drop_error() -> TestResult<()> { Ok(()) } +/// Checks that a sender gets `ReceiverClosed` if the remote receiver is dropped, as for a local one. +#[tokio::test] +async fn mpsc_receiver_dropped() -> TestResult<()> { + let (server, client, server_addr) = create_connected_endpoints()?; + let server = tokio::spawn(async move { + let conn = server.accept().await.unwrap().await?; + let (_, recv) = conn.accept_bi().await?; + drop(Receiver::>::from(recv)); + // keep the connection, so that the stop reaches the client + conn.closed().await; + TestResult::Ok(()) + }); + let conn = client.connect(server_addr, "localhost")?.await?; + let (send, _) = conn.open_bi().await?; + let send = mpsc::Sender::>::from(send); + let res = timeout(Duration::from_secs(5), async { + loop { + if let Err(err) = send.send(vec![1, 2, 3]).await { + break err; + } + } + }) + .await?; + assert!( + matches!(res, SendError::ReceiverClosed { .. }), + "expected ReceiverClosed, got {res:?}" + ); + conn.close(0u32.into(), b""); + server.await??; + Ok(()) +} + +/// Check that `ReceiverClosed` works well with the `Handler` as well. +#[tokio::test] +async fn mpsc_receiver_dropped_proto() -> TestResult<()> { + #[rpc_requests(message = ProtoMessage)] + #[derive(Debug, Serialize, Deserialize)] + enum Proto { + #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver)] + #[wrap(ClientStream)] + ClientStream(()), + } + let (server, client, server_addr) = create_connected_endpoints()?; + let server = tokio::spawn(async move { + let conn = server.accept().await.unwrap().await.unwrap(); + Handler::::sequential(|msg| async move { + match msg { + ProtoMessage::ClientStream(msg) => { + let WithChannels { tx, mut rx, .. } = msg; + tx.send(()).await.unwrap(); + let n = rx.recv().await.unwrap().unwrap(); + assert_eq!(n, 1); + drop(rx); + } + } + Ok(()) + }) + .handle_connection(&conn) + .await + .unwrap(); + }); + let client = Client::::noq(client, server_addr); + let (tx, rx) = client.client_streaming(ClientStream(()), 4).await.unwrap(); + assert!(matches!(rx.await.unwrap(), ())); + + // First send works. + let res = tx.send(1).await; + assert!(matches!(res, Ok(()))); + + // As soon as the Stopped frame arrives, our sends fail. + n0_future::time::timeout(Duration::from_secs(1), async { + loop { + let res = tx.send(2).await; + if matches!(res, Err(SendError::ReceiverClosed { .. })) { + break; + } + } + }) + .await + .unwrap(); + + drop(client); + server.await.unwrap(); + Ok(()) +} + async fn vec_receiver(server: Endpoint) -> Result<(), RecvError> { let conn = server .accept()