warp-server/src/main.rs

   1 use futures_util::{FutureExt, StreamExt, SinkExt};
   2 use warp::Filter;
   3
   4 // GET /hello/warp => 200 OK with body "Hello, warp!"
   5 async fn hello(name: String, agent: String) -> Result<impl warp::Reply, warp::Rejection> {
   6     Ok(format!("Hello, {} from {}!", name, agent))
   7 }
   8
   9 // websocat ws://127.0.0.1:3030/ws-echo
  10 async fn ws_echo_connected(websocket: warp::ws::WebSocket) {
  11     // echo all messages back
  12     let (tx, rx) = websocket.split();
  13     rx.forward(tx).map(|result| {
  14         if let Err(e) = result {
  15             log::warn!("websocket error: {:?}", e);
  16         }
  17     }).await;
  18 }
  19
  20 // websocat ws://127.0.0.1:3030/ws-rev
  21 async fn ws_rev_connected(websocket: warp::ws::WebSocket) {
  22     // echo all messages back
  23     tokio::task::spawn(async {
  24         let (mut tx, mut rx) = websocket.split();
  25         while let Some(message) = rx.next().await {
  26             let msg = match message {
  27                 Ok(msg) =>  msg,
  28                 Err(e) => {
  29                     log::error!("websocket error: {}", e);
  30                     break;
  31                 }
  32             };
  33             log::info!("ws_rev_connected: got message: {:?}", msg);
  34
  35             if msg.is_close() {
  36                 break;
  37             }
  38             if msg.is_text() {
  39                 let text = msg.to_str().unwrap();
  40                 let rev = text.chars().rev().collect::<String>();
  41                 if let Err(e) = tx.send(warp::ws::Message::text(rev)).await {
  42                     // disconnected
  43                     log::info!("peer disconnected: {}", e);
  44                     break;
  45                 }
  46             }
  47             if msg.is_binary() {
  48                 let mut rev = msg.into_bytes();
  49                 rev.reverse();
  50                 if let Err(e) = tx.send(warp::ws::Message::binary(rev)).await {
  51                     // disconnected
  52                     log::info!("peer disconnected: {}", e);
  53                     break;
  54                 }
  55             }
  56         }
  57         log::info!("ws_rev_connected ended");
  58     });
  59 }
  60
  61 #[tokio::main]
  62 async fn main() {
  63     env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
  64
  65     let hello = warp::path!("hello" / String)
  66         .and(warp::header::<String>("user-agent"))
  67         .and_then(hello);
  68
  69     let ws_echo = warp::path("ws-echo")
  70         .and(warp::ws())
  71         .map(|ws: warp::ws::Ws| { ws.on_upgrade(ws_echo_connected) });
  72
  73     let ws_rev = warp::path("ws-rev")
  74         .and(warp::ws())
  75         .map(|ws: warp::ws::Ws| { ws.on_upgrade(ws_rev_connected) });
  76
  77     let api = hello
  78         .or(ws_echo)
  79         .or(ws_rev)
  80         .with(warp::log("warp-server"));
  81
  82     warp::serve(api)
  83         .run(([127, 0, 0, 1], 3030))
  84         .await;
  85 }