|
| 1 | +use std::{ |
| 2 | + io::ErrorKind, |
| 3 | + net::{Ipv4Addr, SocketAddr, SocketAddrV4}, |
| 4 | + time::Duration, |
| 5 | +}; |
| 6 | + |
| 7 | +use irpc::{ |
| 8 | + channel::{spsc, SendError}, |
| 9 | + util::{make_client_endpoint, make_server_endpoint}, |
| 10 | +}; |
| 11 | +use quinn::Endpoint; |
| 12 | +use testresult::TestResult; |
| 13 | +use tokio::time::timeout; |
| 14 | + |
| 15 | +fn create_connected_endpoints() -> TestResult<(Endpoint, Endpoint, SocketAddr)> { |
| 16 | + let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into(); |
| 17 | + let (server, cert) = make_server_endpoint(addr)?; |
| 18 | + let client = make_client_endpoint(addr, &[cert.as_slice()])?; |
| 19 | + let port = server.local_addr()?.port(); |
| 20 | + let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into(); |
| 21 | + Ok((server, client, server_addr)) |
| 22 | +} |
| 23 | + |
| 24 | +/// Checks that all clones of a `Sender` will get the closed signal as soon as |
| 25 | +/// a send fails with an io error. |
| 26 | +#[tokio::test] |
| 27 | +async fn mpsc_sender_clone_closed_error() -> TestResult<()> { |
| 28 | + tracing_subscriber::fmt::try_init().ok(); |
| 29 | + let (server, client, server_addr) = create_connected_endpoints()?; |
| 30 | + // accept a single bidi stream on a single connection, then immediately stop it |
| 31 | + let server = tokio::spawn(async move { |
| 32 | + let conn = server.accept().await.unwrap().await?; |
| 33 | + let (_, mut recv) = conn.accept_bi().await?; |
| 34 | + recv.stop(1u8.into())?; |
| 35 | + TestResult::Ok(()) |
| 36 | + }); |
| 37 | + let conn = client.connect(server_addr, "localhost")?.await?; |
| 38 | + let (send, _) = conn.open_bi().await?; |
| 39 | + let send1 = spsc::Sender::<Vec<u8>>::from(send); |
| 40 | + let send2 = send1.clone(); |
| 41 | + let send3 = send1.clone(); |
| 42 | + let second_client = tokio::spawn(async move { |
| 43 | + send2.closed().await; |
| 44 | + }); |
| 45 | + let third_client = tokio::spawn(async move { |
| 46 | + // this should fail with an io error, since the stream was stopped |
| 47 | + loop { |
| 48 | + match send3.send(vec![1, 2, 3]).await { |
| 49 | + Err(SendError::Io(e)) if e.kind() == ErrorKind::BrokenPipe => break, |
| 50 | + _ => {} |
| 51 | + }; |
| 52 | + } |
| 53 | + }); |
| 54 | + // send until we get an error because the remote side stopped the stream |
| 55 | + while send1.send(vec![1, 2, 3]).await.is_ok() {} |
| 56 | + match send1.send(vec![4, 5, 6]).await { |
| 57 | + Err(SendError::Io(e)) if e.kind() == ErrorKind::BrokenPipe => {} |
| 58 | + e => panic!("Expected SendError::Io with kind BrokenPipe, got {:?}", e), |
| 59 | + }; |
| 60 | + // check that closed signal was received by the second sender |
| 61 | + second_client.await?; |
| 62 | + // check that the third sender will get the right kind of io error eventually |
| 63 | + third_client.await?; |
| 64 | + // server should finish without errors |
| 65 | + server.await??; |
| 66 | + Ok(()) |
| 67 | +} |
| 68 | + |
| 69 | +/// Checks that all clones of a `Sender` will get the closed signal as soon as |
| 70 | +/// a send future gets dropped before completing. |
| 71 | +#[tokio::test] |
| 72 | +async fn mpsc_sender_clone_drop_error() -> TestResult<()> { |
| 73 | + let (server, client, server_addr) = create_connected_endpoints()?; |
| 74 | + // accept a single bidi stream on a single connection, then read indefinitely |
| 75 | + // until we get an error or the stream is finished |
| 76 | + let server = tokio::spawn(async move { |
| 77 | + let conn = server.accept().await.unwrap().await?; |
| 78 | + let (_, mut recv) = conn.accept_bi().await?; |
| 79 | + let mut buf = vec![0u8; 1024]; |
| 80 | + while let Ok(Some(_)) = recv.read(&mut buf).await {} |
| 81 | + TestResult::Ok(()) |
| 82 | + }); |
| 83 | + let conn = client.connect(server_addr, "localhost")?.await?; |
| 84 | + let (send, _) = conn.open_bi().await?; |
| 85 | + let send1 = spsc::Sender::<Vec<u8>>::from(send); |
| 86 | + let send2 = send1.clone(); |
| 87 | + let send3 = send1.clone(); |
| 88 | + let second_client = tokio::spawn(async move { |
| 89 | + send2.closed().await; |
| 90 | + }); |
| 91 | + let third_client = tokio::spawn(async move { |
| 92 | + // this should fail with an io error, since the stream was stopped |
| 93 | + loop { |
| 94 | + match send3.send(vec![1, 2, 3]).await { |
| 95 | + Err(SendError::Io(e)) if e.kind() == ErrorKind::BrokenPipe => break, |
| 96 | + _ => {} |
| 97 | + }; |
| 98 | + } |
| 99 | + }); |
| 100 | + // send a lot of data with a tiny timeout, this will cause the send future to be dropped |
| 101 | + loop { |
| 102 | + let send_future = send1.send(vec![0u8; 1024 * 1024]); |
| 103 | + // not sure if there is a better way. I want to poll the future a few times so it has time to |
| 104 | + // start sending, but don't want to give it enough time to complete. |
| 105 | + // I don't think now_or_never would work, since it wouldn't have time to start sending |
| 106 | + if timeout(Duration::from_micros(1), send_future) |
| 107 | + .await |
| 108 | + .is_err() |
| 109 | + { |
| 110 | + break; |
| 111 | + } |
| 112 | + } |
| 113 | + server.await??; |
| 114 | + second_client.await?; |
| 115 | + third_client.await?; |
| 116 | + Ok(()) |
| 117 | +} |
0 commit comments