xwt_tests/tests/
multiple_streams.rs

1//! This test ensures that multiple streams can be open on a single session
2//! at the same time, and that the data on them is not mixed up.
3
4use xwt_core::prelude::*;
5
6/// The amount of the streams to open.
7const STREAMS: usize = 4;
8
9/// Compute the payload to send over the stream with the given index.
10///
11/// The payloads have different lengths, so that a stream that responds with
12/// the data of another stream is detected even if the reads are aligned.
13fn payload(index: usize) -> Vec<u8> {
14    let mut payload = format!("stream-{index}").into_bytes();
15    payload.extend(std::iter::repeat_n(b'.', index));
16    payload
17}
18
19#[derive(Debug, thiserror::Error)]
20pub enum Error<Endpoint>
21where
22    Endpoint: xwt_core::endpoint::Connect + std::fmt::Debug,
23    Endpoint::Connecting: std::fmt::Debug,
24    ConnectSessionFor<Endpoint>: xwt_core::session::stream::OpenBi + std::fmt::Debug,
25{
26    #[error("connect: {0}")]
27    Connect(#[source] xwt_error::Connect<Endpoint>),
28    #[error("open: {0}")]
29    Open(#[source] xwt_error::OpenBi<ConnectSessionFor<Endpoint>>),
30    #[error("send: {0}")]
31    Send(#[source] WriteErrorFor<SendStreamFor<ConnectSessionFor<Endpoint>>>),
32    #[error("recv: {0}")]
33    Recv(#[source] ReadErrorFor<RecvStreamFor<ConnectSessionFor<Endpoint>>>),
34    #[error("bad data at stream {index}")]
35    BadData { index: usize, data: Vec<u8> },
36}
37
38pub async fn run<Endpoint>(endpoint: Endpoint, url: &str) -> Result<(), Error<Endpoint>>
39where
40    Endpoint: xwt_core::endpoint::Connect + std::fmt::Debug,
41    Endpoint::Connecting: std::fmt::Debug,
42    ConnectSessionFor<Endpoint>: xwt_core::session::stream::OpenBi + std::fmt::Debug,
43{
44    let session = crate::utils::connect(&endpoint, url)
45        .await
46        .map_err(Error::Connect)?;
47
48    let mut streams = Vec::with_capacity(STREAMS);
49
50    for index in 0..STREAMS {
51        let (mut send_stream, recv_stream) =
52            crate::utils::open_bi(&session).await.map_err(Error::Open)?;
53
54        let expected = payload(index);
55
56        let mut to_write = &expected[..];
57        loop {
58            let written = send_stream.write(to_write).await.map_err(Error::Send)?;
59            let written = written.get();
60            to_write = &to_write[written..];
61            if to_write.is_empty() {
62                break;
63            }
64        }
65
66        // The send streams are kept around until the end of the test, as
67        // dropping them can abort the whole stream.
68        streams.push((index, send_stream, recv_stream, expected));
69    }
70
71    // Read the responses in the reverse order to ensure the streams are
72    // independent of one another rather than being a single queue.
73    for (index, _send_stream, mut recv_stream, expected) in streams.into_iter().rev() {
74        let mut read_buf = vec![0u8; expected.len()];
75
76        let mut filled = 0;
77        while filled < expected.len() {
78            let read = recv_stream
79                .read(&mut read_buf[filled..])
80                .await
81                .map_err(Error::Recv)?;
82            filled += read.get();
83        }
84
85        if read_buf != expected {
86            return Err(Error::BadData {
87                index,
88                data: read_buf,
89            });
90        }
91    }
92
93    Ok(())
94}
xwt_tests/tests/multiple_streams.rs

xwt_tests/tests/
multiple_streams.rs
