use reqwest::Client;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use crate::errors::EngineResult;
use super::engine::EngineConfig;
use super::engine::IpFamily;
use super::event::{emit, EngineEvent};
use super::models::{Bufferbloat, IpVersion, LatencyStats, Server, TestPhase, TestReport};
use super::network::{download, icmp, loaded_latency, ping, upload};
pub(crate) async fn run_sequence(
client: &Client,
server: &Server,
config: &EngineConfig,
events: &mpsc::Sender<EngineEvent>,
cancel: &CancellationToken,
) -> EngineResult<TestReport> {
emit(
events,
EngineEvent::PhaseStarted {
phase: TestPhase::Ping,
},
)
.await?;
let (latency, icmp_loss) = tokio::join!(
ping::measure_latency(client, &server.endpoints.ping, &config.ping, events, cancel),
measure_icmp_loss(server, cancel),
);
let mut latency: LatencyStats = latency?;
if let Some(loss) = icmp_loss {
latency.packet_loss_pct = loss;
}
emit(events, EngineEvent::PingFinished { stats: latency }).await?;
emit(
events,
EngineEvent::PhaseStarted {
phase: TestPhase::Download,
},
)
.await?;
lead_in(config, cancel).await?;
let sampler_stop = cancel.child_token();
let (download_stats, loaded_down_ms) = tokio::join!(
async {
let result = download::run_download(
client,
&server.endpoints.download,
&config.transfer,
events,
cancel,
)
.await;
sampler_stop.cancel();
result
},
loaded_latency::sample_until_stopped(client, &server.endpoints.ping, &sampler_stop),
);
let download_stats = download_stats?;
emit(
events,
EngineEvent::TransferFinished {
phase: TestPhase::Download,
stats: download_stats,
},
)
.await?;
emit(
events,
EngineEvent::PhaseStarted {
phase: TestPhase::Upload,
},
)
.await?;
lead_in(config, cancel).await?;
let sampler_stop = cancel.child_token();
let (upload_stats, loaded_up_ms) = tokio::join!(
async {
let result = upload::run_upload(
client,
&server.endpoints.upload,
&config.transfer,
events,
cancel,
)
.await;
sampler_stop.cancel();
result
},
loaded_latency::sample_until_stopped(client, &server.endpoints.ping, &sampler_stop),
);
let upload_stats = upload_stats?;
emit(
events,
EngineEvent::TransferFinished {
phase: TestPhase::Upload,
stats: upload_stats,
},
)
.await?;
let bufferbloat = match (loaded_down_ms, loaded_up_ms) {
(Some(down), Some(up)) => Some(Bufferbloat::new(latency.average_ms, down, up)),
_ => None,
};
let ip_version = config.ip_family.map(|f| match f {
IpFamily::V4 => IpVersion::V4,
IpFamily::V6 => IpVersion::V6,
});
Ok(TestReport {
server_name: server.name.clone(),
latency,
download: download_stats,
upload: upload_stats,
bufferbloat,
ip_version,
})
}
async fn measure_icmp_loss(server: &Server, cancel: &CancellationToken) -> Option<f64> {
let host = icmp::host_of_url(&server.endpoints.ping)?;
icmp::measure_loss(&host, cancel).await
}
async fn lead_in(config: &EngineConfig, cancel: &CancellationToken) -> EngineResult<()> {
if config.transfer.lead_in.is_zero() {
return Ok(());
}
tokio::select! {
() = cancel.cancelled() => Err(crate::errors::EngineError::Cancelled),
() = tokio::time::sleep(config.transfer.lead_in) => Ok(()),
}
}