Struct Server

pub struct Server { /* private fields */ }

表示一个TCP服务器

Server 是框架的主要入口点，负责监听TCP连接并处理客户端请求。 它使用 Router 来分发请求，使用 Connection 来管理客户端连接。

Implementations

impl Server

pub fn new(addr: &str, router: Arc<dyn Router + Send + Sync>) -> Self

创建一个新的服务器实例

参数

• addr - 服务器监听的地址，格式为 “IP:端口”
• router - 路由器实例，用于分发请求到对应的处理函数

返回值

返回一个新的 Server 实例

Examples found in repository

examples/benchmark_server.rs (line 87)

async fn run_server() -> Result<(), Box<dyn std::error::Error>> {
    // 创建关闭通道
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    // 创建路由器并注册回显处理函数
    let router = Arc::new(DefaultRouter::new());
    let router_clone = router.clone();

    // 计数器，用于统计处理的请求数
    let request_counter = Arc::new(AtomicUsize::new(0));
    let counter_clone = request_counter.clone();

    // 注册高性能回显处理函数 - 不打印日志，直接返回
    router_clone.add_route(1, move |req| {
        counter_clone.fetch_add(1, Ordering::Relaxed);
        Response::new(req.msg_id(), req.data().to_vec())
    });

    // 启动服务器
    let server_addr = "127.0.0.1:8888";
    let server = Server::new(server_addr, router);
    println!("[Server] 基准测试服务器启动在 {}", server_addr);

    // 启动统计任务
    let stats_handle = tokio::spawn(async move {
        let mut last_count = 0;
        let mut last_time = Instant::now();

        loop {
            sleep(Duration::from_secs(1)).await;
            let current_count = request_counter.load(Ordering::Relaxed);
            let current_time = Instant::now();
            let elapsed = current_time.duration_since(last_time).as_secs_f64();

            let rps = (current_count - last_count) as f64 / elapsed;
            println!(
                "[Stats] 当前RPS: {:.2} req/s, 总请求数: {}",
                rps, current_count
            );

            last_count = current_count;
            last_time = current_time;
        }
    });

    // 启动服务器并等待Ctrl+C信号
    let server_handle = tokio::spawn(async move {
        if let Err(e) = server.run(shutdown_rx).await {
            eprintln!("[Server] 运行时错误: {}", e);
        }
    });

    println!("[Server] 按 Ctrl+C 停止服务器...");
    tokio::signal::ctrl_c().await?;
    println!("[Server] 接收到停止信号，正在关闭...");

    // 发送关闭信号
    let _ = shutdown_tx.send(());

    // 等待服务器和统计任务完成
    let _ = server_handle.await;
    stats_handle.abort();

    println!("[Server] 服务器已关闭");
    Ok(())
}

examples/echo_server_v1.rs (line 47)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ========================================
    // 1. 创建关闭通道：用于外部控制服务器生命周期
    // ========================================
    // 当 shutdown_tx 被 drop 或 send(()) 时，shutdown_rx 将完成
    // server.run() 中通过 tokio::select! 监听该信号，实现优雅退出
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    // ========================================
    // 2. 创建并配置路由器
    // ========================================
    let router = Arc::new(DefaultRouter::new());

    // 注册 msg_id = 1 的回显处理函数
    let router_clone = router.clone();
    router_clone.add_route(1, |req| {
        println!("Received echo request: {:?}", req.data());
        Response::new(req.msg_id(), req.data().to_vec()) // 原样返回
    });

    // ========================================
    // 3. 启动服务器（异步任务）
    // ========================================
    let server = Server::new("127.0.0.1:8000", router);
    let server_handle = tokio::spawn(async move {
        if let Err(e) = server.run(shutdown_rx).await {
            eprintln!("[Zerust] Server runtime error: {}", e);
        }
    });

    // ========================================
    // 4. 等待服务器就绪（端口探测）
    // ========================================
    // 替代 sleep()，更可靠：最多等待 5 秒，每 10ms 尝试一次连接
    if let Err(_) = wait_for_server(8000, Duration::from_secs(5)).await {
        eprintln!("[Client] Failed to connect to server within 5 seconds.");
        return Err("Server did not start in time".into());
    }
    println!("[Client] Server is ready. Proceeding with test...");

    // ========================================
    // 5. 运行客户端测试
    // ========================================
    match client().await {
        Ok(()) => println!("✅ Client finished successfully."),
        Err(e) => eprintln!("❌ Client error: {}", e),
    }

    // ========================================
    // 6. 发送关闭信号
    // ========================================
    // 客户端完成，通知服务器关闭
    let _ = shutdown_tx.send(());
    println!("[Main] Shutdown signal sent.");

    // ========================================
    // 7. 等待服务器完全停止
    // ========================================
    // 确保 server.run() 任务完全结束，避免资源泄漏
    let _ = server_handle.await;

    println!("🎉 Program exited gracefully.");
    Ok(())
}

pub async fn run(&self, shutdown: Receiver<()>) -> Result<(), ZerustError>

启动服务器并监听指定地址的TCP连接

该函数会绑定到配置的地址并开始监听TCP连接，对于每个传入的连接， 都会创建一个异步任务来处理请求。如果在监听过程中发生IO错误， 函数会立即返回错误。

参数

• shutdown: 接收关闭信号的通道。当发送端被 drop 或发送消息时，服务器将关闭。

返回值

• Ok(()) - 服务器正常启动并运行
• Err(ZerustError) - 服务器启动或运行过程中发生错误

Examples found in repository

examples/benchmark_server.rs (line 114)

async fn run_server() -> Result<(), Box<dyn std::error::Error>> {
    // 创建关闭通道
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    // 创建路由器并注册回显处理函数
    let router = Arc::new(DefaultRouter::new());
    let router_clone = router.clone();

    // 计数器，用于统计处理的请求数
    let request_counter = Arc::new(AtomicUsize::new(0));
    let counter_clone = request_counter.clone();

    // 注册高性能回显处理函数 - 不打印日志，直接返回
    router_clone.add_route(1, move |req| {
        counter_clone.fetch_add(1, Ordering::Relaxed);
        Response::new(req.msg_id(), req.data().to_vec())
    });

    // 启动服务器
    let server_addr = "127.0.0.1:8888";
    let server = Server::new(server_addr, router);
    println!("[Server] 基准测试服务器启动在 {}", server_addr);

    // 启动统计任务
    let stats_handle = tokio::spawn(async move {
        let mut last_count = 0;
        let mut last_time = Instant::now();

        loop {
            sleep(Duration::from_secs(1)).await;
            let current_count = request_counter.load(Ordering::Relaxed);
            let current_time = Instant::now();
            let elapsed = current_time.duration_since(last_time).as_secs_f64();

            let rps = (current_count - last_count) as f64 / elapsed;
            println!(
                "[Stats] 当前RPS: {:.2} req/s, 总请求数: {}",
                rps, current_count
            );

            last_count = current_count;
            last_time = current_time;
        }
    });

    // 启动服务器并等待Ctrl+C信号
    let server_handle = tokio::spawn(async move {
        if let Err(e) = server.run(shutdown_rx).await {
            eprintln!("[Server] 运行时错误: {}", e);
        }
    });

    println!("[Server] 按 Ctrl+C 停止服务器...");
    tokio::signal::ctrl_c().await?;
    println!("[Server] 接收到停止信号，正在关闭...");

    // 发送关闭信号
    let _ = shutdown_tx.send(());

    // 等待服务器和统计任务完成
    let _ = server_handle.await;
    stats_handle.abort();

    println!("[Server] 服务器已关闭");
    Ok(())
}

examples/echo_server_v1.rs (line 49)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ========================================
    // 1. 创建关闭通道：用于外部控制服务器生命周期
    // ========================================
    // 当 shutdown_tx 被 drop 或 send(()) 时，shutdown_rx 将完成
    // server.run() 中通过 tokio::select! 监听该信号，实现优雅退出
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    // ========================================
    // 2. 创建并配置路由器
    // ========================================
    let router = Arc::new(DefaultRouter::new());

    // 注册 msg_id = 1 的回显处理函数
    let router_clone = router.clone();
    router_clone.add_route(1, |req| {
        println!("Received echo request: {:?}", req.data());
        Response::new(req.msg_id(), req.data().to_vec()) // 原样返回
    });

    // ========================================
    // 3. 启动服务器（异步任务）
    // ========================================
    let server = Server::new("127.0.0.1:8000", router);
    let server_handle = tokio::spawn(async move {
        if let Err(e) = server.run(shutdown_rx).await {
            eprintln!("[Zerust] Server runtime error: {}", e);
        }
    });

    // ========================================
    // 4. 等待服务器就绪（端口探测）
    // ========================================
    // 替代 sleep()，更可靠：最多等待 5 秒，每 10ms 尝试一次连接
    if let Err(_) = wait_for_server(8000, Duration::from_secs(5)).await {
        eprintln!("[Client] Failed to connect to server within 5 seconds.");
        return Err("Server did not start in time".into());
    }
    println!("[Client] Server is ready. Proceeding with test...");

    // ========================================
    // 5. 运行客户端测试
    // ========================================
    match client().await {
        Ok(()) => println!("✅ Client finished successfully."),
        Err(e) => eprintln!("❌ Client error: {}", e),
    }

    // ========================================
    // 6. 发送关闭信号
    // ========================================
    // 客户端完成，通知服务器关闭
    let _ = shutdown_tx.send(());
    println!("[Main] Shutdown signal sent.");

    // ========================================
    // 7. 等待服务器完全停止
    // ========================================
    // 确保 server.run() 任务完全结束，避免资源泄漏
    let _ = server_handle.await;

    println!("🎉 Program exited gracefully.");
    Ok(())
}