xwt_tests/tests/
abort_bi_send_stream.rs

1//! This test ensures that aborting the write side of a stream propagates
2//! the error code to the peer.
3//!
4//! The server echoes the data we send first - so that we only abort a stream
5//! that is already established at the server side - and then reports the error
6//! code it has observed over a unidirectional stream.
7
8use xwt_core::prelude::*;
9
10#[derive(Debug, thiserror::Error)]
11pub enum Error<Endpoint>
12where
13    Endpoint: xwt_core::endpoint::Connect + std::fmt::Debug,
14    Endpoint::Connecting: std::fmt::Debug,
15    ConnectSessionFor<Endpoint>:
16        xwt_core::session::stream::OpenBi + xwt_core::session::stream::AcceptUni + std::fmt::Debug,
17{
18    #[error("connect: {0}")]
19    Connect(#[source] xwt_error::Connect<Endpoint>),
20    #[error("open: {0}")]
21    Open(#[source] xwt_error::OpenBi<ConnectSessionFor<Endpoint>>),
22    #[error("send: {0}")]
23    Send(#[source] WriteErrorFor<SendStreamFor<ConnectSessionFor<Endpoint>>>),
24    #[error("recv: {0}")]
25    Recv(#[source] ReadErrorFor<RecvStreamFor<ConnectSessionFor<Endpoint>>>),
26    #[error("write stream abort: {0}")]
27    WriteStreamAbort(#[source] WriteAbortErrorFor<SendStreamFor<ConnectSessionFor<Endpoint>>>),
28    #[error("accept uni stream: {0}")]
29    AcceptUniStream(#[source] UniStreamAcceptErrorFor<ConnectSessionFor<Endpoint>>),
30    #[error("error code mismatch: got code {0}")]
31    ErrorCodeMismatch(xwt_core::stream::ErrorCode),
32}
33
34pub async fn run<Endpoint>(
35    endpoint: Endpoint,
36    url: &str,
37    error_code: xwt_core::stream::ErrorCode,
38) -> Result<(), Error<Endpoint>>
39where
40    Endpoint: xwt_core::endpoint::Connect + std::fmt::Debug,
41    Endpoint::Connecting: std::fmt::Debug,
42    ConnectSessionFor<Endpoint>:
43        xwt_core::session::stream::OpenBi + xwt_core::session::stream::AcceptUni + std::fmt::Debug,
44{
45    let session = crate::utils::connect(&endpoint, url)
46        .await
47        .map_err(Error::Connect)?;
48
49    let (mut send_stream, mut recv_stream) =
50        crate::utils::open_bi(&session).await.map_err(Error::Open)?;
51
52    let mut to_write = &b"ping"[..];
53    loop {
54        let written = send_stream.write(to_write).await.map_err(Error::Send)?;
55        let written = written.get();
56        to_write = &to_write[written..];
57        if to_write.is_empty() {
58            break;
59        }
60    }
61
62    // Wait for the echo to ensure the server has this stream before we abort
63    // it - a stream that is reset before it reaches the peer might never be
64    // observed there at all.
65    let mut echo_buf = [0u8; 4];
66    let mut filled = 0;
67    while filled < echo_buf.len() {
68        let read = recv_stream
69            .read(&mut echo_buf[filled..])
70            .await
71            .map_err(Error::Recv)?;
72        filled += read.get();
73    }
74
75    send_stream
76        .abort(error_code)
77        .await
78        .map_err(Error::WriteStreamAbort)?;
79
80    let mut report_stream = session.accept_uni().await.map_err(Error::AcceptUniStream)?;
81
82    let mut report_buf = [0u8; 4];
83    let mut filled = 0;
84    while filled < report_buf.len() {
85        let read = report_stream
86            .read(&mut report_buf[filled..])
87            .await
88            .map_err(Error::Recv)?;
89        filled += read.get();
90    }
91
92    let observed_error_code = xwt_core::stream::ErrorCode::from_be_bytes(report_buf);
93
94    if observed_error_code != error_code {
95        return Err(Error::ErrorCodeMismatch(observed_error_code));
96    }
97
98    Ok(())
99}
xwt_tests/tests/abort_bi_send_stream.rs

xwt_tests/tests/
abort_bi_send_stream.rs
