-mod handlers {
- use futures_util::{FutureExt, StreamExt, SinkExt};
+use futures_util::{FutureExt, StreamExt, SinkExt};
+use warp::Filter;
- // GET /hello/warp => 200 OK with body "Hello, warp!"
- pub async fn hello(name: String, agent: String) -> Result<impl warp::Reply, warp::Rejection> {
- Ok(format!("Hello, {} from {}!", name, agent))
- }
+// GET /hello/warp => 200 OK with body "Hello, warp!"
+async fn hello(name: String, agent: String) -> Result<impl warp::Reply, warp::Rejection> {
+ Ok(format!("Hello, {} from {}!", name, agent))
+}
- // websocat ws://127.0.0.1:3030/ws-echo
- pub async fn ws_echo_connected(websocket: warp::ws::WebSocket) {
- // echo all messages back
- let (tx, rx) = websocket.split();
- rx.forward(tx).map(|result| {
- if let Err(e) = result {
- log::warn!("websocket error: {:?}", e);
- }
- }).await;
- }
+// websocat ws://127.0.0.1:3030/ws-echo
+async fn ws_echo_connected(websocket: warp::ws::WebSocket) {
+ // echo all messages back
+ let (tx, rx) = websocket.split();
+ rx.forward(tx).map(|result| {
+ if let Err(e) = result {
+ log::warn!("websocket error: {:?}", e);
+ }
+ }).await;
+}
- // websocat ws://127.0.0.1:3030/ws-rev
- pub async fn ws_rev_connected(websocket: warp::ws::WebSocket) {
- // echo all messages back
- tokio::task::spawn(async {
- let (mut tx, mut rx) = websocket.split();
- while let Some(message) = rx.next().await {
- let msg = match message {
- Ok(msg) => msg,
- Err(e) => {
- log::error!("websocket error: {}", e);
- break;
- }
- };
- log::info!("ws_rev_connected: got message: {:?}", msg);
-
- if msg.is_close() {
+// websocat ws://127.0.0.1:3030/ws-rev
+async fn ws_rev_connected(websocket: warp::ws::WebSocket) {
+ // echo all messages back
+ tokio::task::spawn(async {
+ let (mut tx, mut rx) = websocket.split();
+ while let Some(message) = rx.next().await {
+ let msg = match message {
+ Ok(msg) => msg,
+ Err(e) => {
+ log::error!("websocket error: {}", e);
break;
}
- if msg.is_text() {
- let text = msg.to_str().unwrap();
- let rev = text.chars().rev().collect::<String>();
- if let Err(e) = tx.send(warp::ws::Message::text(rev)).await {
- // disconnected
- log::info!("peer disconnected: {}", e);
- break;
- }
+ };
+ log::info!("ws_rev_connected: got message: {:?}", msg);
+
+ if msg.is_close() {
+ break;
+ }
+ if msg.is_text() {
+ let text = msg.to_str().unwrap();
+ let rev = text.chars().rev().collect::<String>();
+ if let Err(e) = tx.send(warp::ws::Message::text(rev)).await {
+ // disconnected
+ log::info!("peer disconnected: {}", e);
+ break;
}
- if msg.is_binary() {
- let mut rev = msg.into_bytes();
- rev.reverse();
- if let Err(e) = tx.send(warp::ws::Message::binary(rev)).await {
- // disconnected
- log::info!("peer disconnected: {}", e);
- break;
- }
+ }
+ if msg.is_binary() {
+ let mut rev = msg.into_bytes();
+ rev.reverse();
+ if let Err(e) = tx.send(warp::ws::Message::binary(rev)).await {
+ // disconnected
+ log::info!("peer disconnected: {}", e);
+ break;
}
}
- log::info!("ws_rev_connected ended");
- });
- }
+ }
+ log::info!("ws_rev_connected ended");
+ });
}
-mod filters {
- use warp::Filter;
- use super::handlers;
+pub fn api() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
+ let hello = warp::path!("hello" / String)
+ .and(warp::header::<String>("user-agent"))
+ .and_then(hello);
- pub fn hello() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
- warp::path!("hello" / String)
- .and(warp::header::<String>("user-agent"))
- .and_then(handlers::hello)
- }
+ let ws_echo = warp::path("ws-echo")
+ .and(warp::ws())
+ .map(|ws: warp::ws::Ws| { ws.on_upgrade(ws_echo_connected) });
- pub fn ws_echo() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
- warp::path("ws-echo")
- .and(warp::ws())
- .map(|ws: warp::ws::Ws| { ws.on_upgrade(handlers::ws_echo_connected) })
- }
-
- pub fn ws_rev() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
- warp::path("ws-rev")
- .and(warp::ws())
- .map(|ws: warp::ws::Ws| { ws.on_upgrade(handlers::ws_rev_connected) })
- }
+ let ws_rev = warp::path("ws-rev")
+ .and(warp::ws())
+ .map(|ws: warp::ws::Ws| { ws.on_upgrade(ws_rev_connected) });
- pub fn api() -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
- hello()
- .or(ws_echo())
- .or(ws_rev())
- .with(warp::log("warp-server"))
- }
+ hello
+ .or(ws_echo)
+ .or(ws_rev)
+ .with(warp::log("warp-server"))
}
#[tokio::main]
async fn main() {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
- warp::serve(filters::api())
+
+ warp::serve(api())
.run(([127, 0, 0, 1], 3030))
.await;
}
+
+#[cfg(test)]
+mod tests {
+ #[tokio::test]
+ async fn test_hello() {
+ let res = warp::test::request()
+ .path("/hello/rust")
+ .header("user-agent", "TestBrowser 0.1")
+ .reply(&super::api())
+ .await;
+ assert_eq!(res.status(), 200);
+ assert_eq!(res.body(), "Hello, rust from TestBrowser 0.1!");
+ }
+
+ #[tokio::test]
+ async fn test_ws_echo() {
+ let mut client = warp::test::ws()
+ .path("/ws-echo")
+ .handshake(super::api())
+ .await
+ .expect("handshake failed");
+
+ // text
+ client.send_text("Hello").await;
+ let reply = client.recv().await.unwrap();
+ assert_eq!(reply.to_str().unwrap(), "Hello");
+
+ // binary
+ let msg: Vec<u8> = vec![42, 99];
+ client.send(warp::ws::Message::binary(msg.clone())).await;
+ let reply = client.recv().await.unwrap();
+ assert_eq!(reply.as_bytes(), &msg);
+
+ //close
+ client.send(warp::ws::Message::close()).await;
+ client.recv_closed().await.unwrap();
+ }
+
+ #[tokio::test]
+ async fn test_ws_rev() {
+ let mut client = warp::test::ws()
+ .path("/ws-rev")
+ .handshake(super::api())
+ .await
+ .expect("handshake failed");
+
+ // text
+ client.send_text("Hello\n").await;
+ let reply = client.recv().await.unwrap();
+ assert_eq!(reply.to_str().unwrap(), "\nolleH");
+
+ // binary
+ let msg: Vec<u8> = vec![42, 99];
+ client.send(warp::ws::Message::binary(msg)).await;
+ let reply = client.recv().await.unwrap();
+ assert_eq!(reply.as_bytes(), vec![99, 42]);
+
+ //close
+ client.send(warp::ws::Message::close()).await;
+ client.recv_closed().await.unwrap();
+ }
+}
projects / learn-rust.git / blobdiff