2

我有一个使用warp. 它实现了一个 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，编译器会抱怨由于期货无法跨线程发送。

如何在拥有路由处理程序本身的同时实现共享应用程序状态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>
4

1 回答 1

6

您通过共享所有权（例如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))
}

这里有第二组clones 因为有两个不同的东西拥有数据：

  1. 处理程序本身（闭包）
  2. 闭包返回的未来（异步代码）

也可以看看：

[dependencies]
warp = "0.3.0"
tokio = { version = "1.2.0", features = ["full"] }
于 2021-02-09T02:54:57.640 回答