我有一个使用 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);
我怀疑这实际上是正确的行为(尽管编译和运行)。
对于一个相对简单的要求来说,这似乎非常丑陋。
最重要的是,在我的路由处理程序中,我需要调用
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>
最佳答案
您通过共享所有权(例如 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
,因为有两个不同的东西拥有数据:
- 处理程序本身(闭包)
- 闭包返回的 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"] }
关于asynchronous - 如何使用 Warp 异步路由实现共享应用程序状态?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/66111599/