如何在Warp框架中全局注入MongoDB连接供API处理器使用

Warp 本身通过 Filter 组合机制实现共享状态注入，mongodb 2.x 版本的Client、Database、Collection实例内部都已经用Arc做了线程安全包装，克隆成本极低，非常适合作为全局共享状态传给各个路由处理函数。

第一步：修改 auth.rs 路由定义，接收共享的 Collection 实例 #

把auth_filter改成接收数据库集合实例作为参数，通过 warp 自带的状态注入能力把实例传到处理链路中：

use serde_json::{json, Value};
use warp::{Filter, reply::Json};
use mongodb::{Collection, bson::doc};

// 接收共享的users集合实例作为入参
pub fn auth_filter(
    users: Collection<bson::Document>
) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
    // 注入集合实例到join路由链路
    let join = warp::path("join")
    .and(warp::get())
    .and(warp::path::end())
    .and(with_collection(users.clone()))
    .and_then(join_handler);

    // 注入集合实例到signin路由链路
    let signin = warp::path("signin")
    .and(warp::post())
    .and(warp::body::json())
    .and(with_collection(users))
    .and_then(signin_handler);

    join.or(signin)
}

// 辅助过滤器，负责把共享实例透传给处理函数
fn with_collection(
    users: Collection<bson::Document>
) -> impl Filter<Extract = (Collection<bson::Document>,), Error = std::convert::Infallible> + Clone {
    warp::any().map(move || users.clone())
}

// 处理函数新增集合参数，可直接操作数据库
async fn join_handler(users: Collection<bson::Document>) -> Result<Json, warp::Rejection> {
    let user = doc! { "password": "1984", "username": "GeorgeOrwell" };
    let insert_result = users.insert_one(user, None).await
        .map_err(|e| warp::reject::custom(DbError(e)))?;
    
    let resp = json!({
        "username":"GeorgeOrwell", 
        "password": "1984", 
        "id": insert_result.inserted_id.as_object_id().unwrap().to_hex()
    });
    Ok(warp::reply::json(&resp))
}

async fn signin_handler(data: Value, users: Collection<bson::Document>) -> Result<Json, warp::Rejection> {
    // 此处可直接调用users集合做账号密码校验逻辑
    Ok(warp::reply::json(&data))
}

// 自定义数据库错误类型，适配warp的拒绝机制
#[derive(Debug)]
struct DbError(mongodb::error::Error);
impl warp::reject::Reject for DbError {}

第二步：修改 main.rs，初始化后传入集合实例 #

把原main函数里的测试插入逻辑移到对应处理函数中，调用auth_filter时传入初始化好的集合实例即可：

mod auth;

use dotenv;
use tokio;
use std::{env, error::Error};
use mongodb::{options::{ClientOptions}, Client};
use warp::Filter;

use crate::auth::auth_filter;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    dotenv::dotenv().ok();
    let client_uri = env::var("MONGODB_URI").expect("You must set the MONGODB_URI .env");
    let options = ClientOptions::parse(&client_uri).await?;
    let client = Client::with_options(options)?;

    // 初始化users集合
    let users = client.database("rusty_db").collection("users");

    // 传入集合实例到路由
    let apis = auth_filter(users);
    let welcome = warp::path::end().map(|| "Welcome to my api");

    let routes = apis.or(welcome);
    warp::serve(routes).run(([127, 0, 0, 1], 3000)).await;

    Ok(())
}

注意事项 #

• 不需要额外给mongodb实例包Arc，官方实现内部已经做了线程安全共享包装，直接clone是浅拷贝，没有性能开销
• 如果后续有其他集合需要共享，按照相同逻辑写对应辅助过滤器注入即可
• 建议补充统一的拒绝处理逻辑，把数据库错误、参数校验错误转换成规范的接口返回格式，不要在处理函数中直接使用unwrap

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