asynchronous - 如何使用 Warp 异步路由实现共享应用程序状态？

Question

我有一个使用 warp 的 Rust 应用程序.它实现了一个 RESTful CRUD API。我需要每个路由处理程序(即最终由 warp 过滤器调用的函数)来访问和(在大多数情况下)改变共享的应用程序状态。

我可以编译它的唯一方法是克隆 Arc<Mutex<State>> 每条路线:

    /* internal_state is loaded from a dump file earlier on and is of type `State` */

    let state: Arc<Mutex<State>> = Arc::new(Mutex::new(internal_state));
    let index_book_state: Arc<Mutex<State>> = state.clone();
    let create_book_state: Arc<Mutex<State>> = state.clone();
    let read_book_state: Arc<Mutex<State>> = state.clone();

    let create_order_state: Arc<Mutex<State>> = state.clone();
    let read_order_state: Arc<Mutex<State>> = state.clone();
    let update_order_state: Arc<Mutex<State>> = state.clone();
    let destroy_order_state: Arc<Mutex<State>> = state.clone();

/* define CRUD routes for order books */
    let book_prefix = warp::path!("book");
    let index_book_route = book_prefix
        .and(warp::get())
        .and(warp::any().map(move || index_book_state.clone()))
        .and_then(handler::index_book_handler);
    let create_book_route = book_prefix
        .and(warp::post())
        .and(warp::body::json())
        .and(warp::any().map(move || create_book_state.clone()))
        .and_then(handler::create_book_handler);
    let read_book_route = warp::path!("book" / String)
        .and(warp::get())
        .and(warp::any().map(move || read_book_state.clone()))
        .and_then(handler::read_book_handler);

    /* define CRUD routes for orders */
    let create_order_route = warp::path!("book" / String)
        .and(warp::post())
        .and(warp::body::json())
        .and(warp::any().map(move || create_order_state.clone()))
        .and_then(handler::create_order_handler);
    let read_order_route = warp::path!("book" / String / "order" / String)
        .and(warp::get())
        .and(warp::any().map(move || read_order_state.clone()))
        .and_then(handler::read_order_handler);
    let update_order_route = warp::path!("book" / String / "order" / String)
        .and(warp::put())
        .and(warp::body::json())
        .and(warp::any().map(move || update_order_state.clone()))
        .and_then(handler::update_order_handler);
    let destroy_order_route = warp::path!("book" / String / "order" / String)
        .and(warp::delete())
        .and(warp::any().map(move || destroy_order_state.clone()))
        .and_then(handler::destroy_order_handler);

    /* aggregate all of our order book routes */
    let book_routes =
        index_book_route.or(create_book_route).or(read_book_route);

    /* aggregate all of our order routes */
    let order_routes = create_order_route
        .or(read_order_route)
        .or(update_order_route)
        .or(destroy_order_route);

    /* aggregate all of our routes */
    let routes = book_routes.or(order_routes);

1. 我怀疑这实际上是正确的行为(尽管编译和运行)。

2. 对于一个相对简单的要求来说，这似乎非常丑陋。

3. 最重要的是，在我的路由处理程序中，我需要调用 async函数，因此需要将处理程序本身标记为 async等。当我将处理程序标记为 async ，编译器提示由于 future 无法跨线程发送。

如何在路由处理程序本身为 async 的同时实现共享应用程序状态?

路由处理程序的签名(它们都是一样的):

/* matches routes like POST `http://example.com/[market]/` */
pub async fn create_order_handler(market: String, request: CreateOrderRequest, state: Arc<Mutex<State>>, rpc_endpoint: String) -> Result<impl Reply, Rejection>

Answer 1

您通过共享所有权(例如 Arc)与线程安全的内部可变性(例如 Mutex、RwLock、或原子):

use std::sync::{Arc, Mutex};
use warp::Filter;

#[tokio::main]
async fn main() {
    let state = Arc::new(Mutex::new(0));

    let market = warp::path!("market" / String).map({
        let state = state.clone();
        move |market| {
            *state.lock().unwrap() += 1;
            format!("Market: {}", market)
        }
    });

    let plaza = warp::path!("plaza" / String).map({
        let state = state.clone();
        move |plaza| {
            let state = *state.lock().unwrap();
            format!("Plaza: {} ({})", plaza, state)
        }
    });

    let routes = market.or(plaza);

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

% curl 127.0.0.1:3030/market/one
Market: one

% curl 127.0.0.1:3030/plaza/one
Plaza: one (1)

要执行异步工作，请使用 Filter::and_then :

use std::{
    convert::Infallible,
    sync::{Arc, Mutex},
};
use warp::Filter;

#[tokio::main]
async fn main() {
    let state = Arc::new(Mutex::new(0));

    let market = warp::path!("market" / String).and_then({
        let state = state.clone();
        move |market| {
            let state = state.clone();
            async move {
                *state.lock().unwrap() += 1;
                Ok::<_, Infallible>(format!("Market: {}", market))
            }
        }
    });

    let plaza = warp::path!("plaza" / String).and_then({
        let state = state.clone();
        move |plaza| {
            let state = state.clone();
            async move {
                let state = *state.lock().unwrap();
                Ok::<_, Infallible>(format!("Plaza: {} ({})", plaza, state))
            }
        }
    });

    let routes = market.or(plaza);

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

这些甚至可以是单独的函数:

use std::{
    convert::Infallible,
    sync::{Arc, Mutex},
};
use warp::Filter;

#[tokio::main]
async fn main() {
    let state = Arc::new(Mutex::new(0));

    let market = warp::path!("market" / String).and_then({
        let state = state.clone();
        move |m| market(m, state.clone())
    });

    let plaza = warp::path!("plaza" / String).and_then({
        let state = state.clone();
        move |p| plaza(p, state.clone())
    });

    let routes = market.or(plaza);

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

type State = Arc<Mutex<i32>>;

async fn market(market: String, state: State) -> Result<String, Infallible> {
    *state.lock().unwrap() += 1;
    Ok::<_, Infallible>(format!("Market: {}", market))
}

async fn plaza(plaza: String, state: State) -> Result<String, Infallible> {
    let state = *state.lock().unwrap();
    Ok::<_, Infallible>(format!("Plaza: {} ({})", plaza, state))
}

这里还有第二组 clone，因为有两个不同的东西拥有数据:

1. 处理程序本身(闭包)
2. 闭包返回的 future (异步代码)

另见:

• Is there another option to share an Arc in multiple closures besides cloning it before each closure?
• Dependency Injection in Rust Warp
• Is there a way to do validation as part of a filter in Warp?

---

[dependencies]
warp = "0.3.0"
tokio = { version = "1.2.0", features = ["full"] }