在Rust warp中实现请求前后中间件：跟踪在途连接

核心思路是借助Rust的Drop特性保证计数器在请求结束（包括异常中断）时总能递减，同时通过正确的Warp过滤器封装避免编译错误。

1. 定义跟踪Guard与指标结构 #

首先实现一个TrackingGuard，在创建时递增计数器，销毁时自动递减：

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::OnceLock;

// 全局共享的HTTP指标实例
#[derive(Clone)]
struct HttpMetrics {
    in_flight: Arc<AtomicU64>,
}

impl HttpMetrics {
    fn new() -> Self {
        HttpMetrics {
            in_flight: Arc::new(AtomicU64::new(0)),
        }
    }

    fn inc_in_flight(&self, path: &str) {
        // 可根据path区分不同接口的在途计数
        self.in_flight.fetch_add(1, Ordering::Relaxed);
        println!("In flight incremented for {}: {}", path, self.in_flight.load(Ordering::Relaxed));
    }

    fn dec_in_flight(&self, path: &str) {
        self.in_flight.fetch_sub(1, Ordering::Relaxed);
        println!("In flight decremented for {}: {}", path, self.in_flight.load(Ordering::Relaxed));
    }
}

static HTTP_METRICS: OnceLock<HttpMetrics> = OnceLock::new();

fn get_metrics() -> &'static HttpMetrics {
    HTTP_METRICS.get_or_init(HttpMetrics::new)
}

// 跟踪Guard，Drop时自动递减计数器
struct TrackingGuard {
    path: String,
}

impl TrackingGuard {
    fn new(path: String) -> Self {
        get_metrics().inc_in_flight(&path);
        TrackingGuard { path }
    }
}

impl Drop for TrackingGuard {
    fn drop(&mut self) {
        get_metrics().dec_in_flight(&self.path);
    }
}

2. 封装Warp过滤器 #

创建一个过滤器封装函数，在请求处理前创建Guard，并确保Guard在整个请求生命周期内存在：

use warp::{Filter, Reply, Rejection, path::FullPath};

/// 为目标过滤器添加在途连接跟踪能力
fn track_in_flight<F>(filter: F) -> impl Filter<Extract = F::Extract, Error = F::Error> + Clone
where
    F: Filter + Clone,
{
    // 先获取请求全路径，用于按接口统计
    warp::path::full()
        .and(filter)
        .wrap_fn(|(path, filter_input)| async move {
            // 创建Guard，此时递增计数器
            let _guard = TrackingGuard::new(path.as_str().to_string());
            // 等待原过滤器处理完成，无论成功/失败/中断，Guard都会自动销毁
            filter_input.await
        })
}

3. 应用到路由中 #

在主函数中使用封装后的过滤器，替代原来的起始/结束指标过滤器：

use std::time::Duration;

#[tokio::main]
async fn main() {
    let hello = warp::path!("hello" / String).and_then(hello);
    let slow = warp::path!("slow").and_then(slow);

    // 用track_in_flight封装所有路由
    let routes = track_in_flight(hello.or(slow));

    warp::serve(routes)
        .run(([127, 0, 0, 1], 8080))
        .await;
}

async fn hello(name: String) -> Result<impl Reply, Rejection> {
    Ok(format!("Hello, {}!", name))
}

async fn slow() -> Result<impl Reply, Rejection> {
    tokio::time::sleep(Duration::from_secs(5)).await;
    Ok(format!("That was slow."))
}

关键说明 #

• 可靠的计数保证：无论请求是正常完成、被拒绝，还是客户端中途断开连接，TrackingGuard都会在异步块结束时被销毁，Drop方法必然执行，确保计数器正确递减。
• 编译错误解决：之前的错误源于错误的函数返回类型（返回Result而非Filter），同时通过path::full()与目标过滤器组合，利用Warp内置的错误类型兼容逻辑，避免了CombineRejection约束问题。

内容的提问来源于stack exchange，提问作者sunside