Struct DefaultRouter

pub struct DefaultRouter { /* private fields */ }

默认路由器实现

使用 DashMap 存储消息ID到处理函数的映射，支持并发访问。 DashMap 是一个线程安全的哈希表，适合在多线程环境中使用。

Implementations

impl DefaultRouter

pub fn new() -> Self

创建一个新的默认路由器实例

返回值

返回一个新的 DefaultRouter 实例，其中包含一个空的路由表

示例

use zerust::{DefaultRouter, Response, Request};
use std::sync::Arc;

// 创建路由器
let router = Arc::new(DefaultRouter::new());

// 添加路由处理
router.add_route(1, |req| {
    println!("处理消息ID为1的请求");
    Response::new(req.msg_id(), b"Hello, World!".to_vec())
});

// 添加另一个路由处理
router.add_route(2, |req| {
    println!("处理消息ID为2的请求");
    Response::new(req.msg_id(), b"Echo: ".iter().chain(req.data().iter()).cloned().collect())
});

Examples found in repository

examples/benchmark_server.rs (line 72)

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 35)

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 fn add_route<F>(&self, msg_id: u32, handler: F)

where F: Fn(&Request) -> Response + Send + Sync + 'static,

添加路由规则

将指定的消息ID与处理函数关联起来，当收到对应消息ID的请求时， 会调用该处理函数生成响应。

参数

• msg_id - 消息ID
• handler - 处理函数，接收请求对象的引用，返回响应对象

类型参数

• F - 处理函数的类型，必须实现 Fn(&Request) -> Response + Send + Sync + 'static
  • 'static 约束确保了闭包捕获的任何数据都拥有所有权或具有 ’static 生命周期， 使得 Handler 可以安全地在程序的整个生命周期内存在

Examples found in repository

examples/benchmark_server.rs (lines 80-83)

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 (lines 39-42)

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(())
}

Trait Implementations

impl Default for DefaultRouter

为 DefaultRouter 实现 Default trait

fn default() -> Self

Returns the “default value” for a type.

impl Router for DefaultRouter

为 DefaultRouter 实现 Router trait

fn handle(&self, req: &Request) -> Response

处理请求并生成响应

根据请求的消息ID查找对应的处理函数，如果找到则调用该函数处理请求， 否则返回一个表示路由未找到的响应。

参数

• req - 请求对象的引用

返回值

返回对应的响应对象