xwt_tests/tests/
large_payload.rs

1//! This test ensures that a payload that is way bigger than a single packet
2//! is transferred over a stream intact and in order.
3//!
4//! The data is sent and read back in rounds, so that the test does not depend
5//! on the flow control window being large enough to hold the whole payload.
6
7use xwt_core::prelude::*;
8
9/// The size of a single round of data.
10const CHUNK_SIZE: usize = 4096;
11
12/// The amount of the rounds of data to send.
13const CHUNKS: usize = 16;
14
15/// Compute the payload byte for the given offset in the stream.
16///
17/// The period is a prime that does not divide [`CHUNK_SIZE`], so that
18/// a reordering of the data at the chunk boundaries alters the payload.
19fn payload_byte(offset: usize) -> u8 {
20    (offset % 251) as u8
21}
22
23#[derive(Debug, thiserror::Error)]
24pub enum Error<Endpoint>
25where
26    Endpoint: xwt_core::endpoint::Connect + std::fmt::Debug,
27    Endpoint::Connecting: std::fmt::Debug,
28    ConnectSessionFor<Endpoint>: xwt_core::session::stream::OpenBi + std::fmt::Debug,
29{
30    #[error("connect: {0}")]
31    Connect(#[source] xwt_error::Connect<Endpoint>),
32    #[error("open: {0}")]
33    Open(#[source] xwt_error::OpenBi<ConnectSessionFor<Endpoint>>),
34    #[error("send: {0}")]
35    Send(#[source] WriteErrorFor<SendStreamFor<ConnectSessionFor<Endpoint>>>),
36    #[error("recv: {0}")]
37    Recv(#[source] ReadErrorFor<RecvStreamFor<ConnectSessionFor<Endpoint>>>),
38    #[error("bad data at offset {offset}")]
39    BadData { offset: usize },
40}
41
42pub async fn run<Endpoint>(endpoint: Endpoint, url: &str) -> Result<(), Error<Endpoint>>
43where
44    Endpoint: xwt_core::endpoint::Connect + std::fmt::Debug,
45    Endpoint::Connecting: std::fmt::Debug,
46    ConnectSessionFor<Endpoint>: xwt_core::session::stream::OpenBi + std::fmt::Debug,
47{
48    let session = crate::utils::connect(&endpoint, url)
49        .await
50        .map_err(Error::Connect)?;
51
52    let (mut send_stream, mut recv_stream) =
53        crate::utils::open_bi(&session).await.map_err(Error::Open)?;
54
55    let mut read_buf = vec![0u8; CHUNK_SIZE];
56
57    for chunk_index in 0..CHUNKS {
58        let offset = chunk_index * CHUNK_SIZE;
59
60        let chunk = (offset..(offset + CHUNK_SIZE))
61            .map(payload_byte)
62            .collect::<Vec<u8>>();
63
64        let mut to_write = &chunk[..];
65        loop {
66            let written = send_stream.write(to_write).await.map_err(Error::Send)?;
67            let written = written.get();
68            to_write = &to_write[written..];
69            if to_write.is_empty() {
70                break;
71            }
72        }
73
74        let mut filled = 0;
75        while filled < CHUNK_SIZE {
76            let read = recv_stream
77                .read(&mut read_buf[filled..])
78                .await
79                .map_err(Error::Recv)?;
80            filled += read.get();
81        }
82
83        if read_buf != chunk {
84            return Err(Error::BadData { offset });
85        }
86    }
87
88    Ok(())
89}
xwt_tests/tests/large_payload.rs

xwt_tests/tests/
large_payload.rs
