strest 0.1.10

Blazing-fast async HTTP load tester in Rust - lock-free design, real-time stats, distributed runs, and optional chart exports for high-load API testing.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

use std::time::Duration;

use tokio::io::BufReader;
use tokio::net::TcpStream;
use tokio::time::timeout;
use tracing::info;

use crate::error::{AppError, AppResult, DistributedError};

use super::super::protocol::{ErrorMessage, WireMessage, read_message, send_message};

pub(super) struct AgentConn {
    pub(super) agent_id: String,
    pub(super) weight: u64,
    pub(super) reader: BufReader<tokio::net::tcp::OwnedReadHalf>,
    pub(super) writer: tokio::net::tcp::OwnedWriteHalf,
}

const AGENT_HELLO_TIMEOUT: Duration = Duration::from_secs(10);

pub(super) async fn accept_agent(
    stream: TcpStream,
    auth_token: Option<&str>,
) -> AppResult<AgentConn> {
    let peer = stream
        .peer_addr()
        .map(|addr| addr.to_string())
        .unwrap_or_else(|_| "<unknown>".to_owned());
    let (read_half, write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);
    let hello = match timeout(AGENT_HELLO_TIMEOUT, read_message(&mut reader)).await {
        Ok(result) => match result? {
            WireMessage::Hello(message) => message,
            WireMessage::Error(message) => {
                return Err(AppError::distributed(DistributedError::Remote {
                    message: message.message,
                }));
            }
            WireMessage::Config(_)
            | WireMessage::Start(_)
            | WireMessage::Stop(_)
            | WireMessage::Heartbeat(_)
            | WireMessage::Report(_)
            | WireMessage::Stream(_) => {
                return Err(AppError::distributed(
                    DistributedError::ExpectedHelloFromAgent,
                ));
            }
        },
        Err(_) => {
            return Err(AppError::distributed(DistributedError::AgentHelloTimeout));
        }
    };

    if let Some(expected) = auth_token {
        let provided = hello.auth_token.as_deref().unwrap_or("");
        if provided != expected {
            let mut writer = write_half;
            send_message(
                &mut writer,
                &WireMessage::Error(ErrorMessage {
                    message: "Invalid auth token.".to_owned(),
                }),
            )
            .await?;
            return Err(AppError::distributed(DistributedError::InvalidAuthToken));
        }
    }

    info!(
        "Accepted agent {} from {} (weight={})",
        hello.agent_id, peer, hello.weight
    );
    Ok(AgentConn {
        agent_id: hello.agent_id,
        weight: hello.weight.max(1),
        reader,
        writer: write_half,
    })
}