Impl websocket

Author: rousan

Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "routerify-websocket"
-version = "0.1.0"
-description = "An websocket extension for Routerify."
+version = "1.0.0"
+description = "The websocket support for the Routerify library."
 homepage = "https://github.com/routerify/routerify-websocket"
 repository = "https://github.com/routerify/routerify-websocket"
 keywords = ["routerify", "hyper-rs", "hyper", "websocket", "ws"]
-categories = ["asynchronous","web-programming","web-programming::http-server"]
+categories = ["asynchronous", "web-programming", "web-programming::websocket"]
 authors = ["Rousan Ali <[email protected]>"]
 readme = "README.md"
 license = "MIT"
@@ -19,17 +19,24 @@ features = ["all"]
 
 [features]
 default = []
-all = []
+all = ["json"]
+json = ["serde", "serde_json"]
 
 [dependencies]
 log = "0.4"
 derive_more = "0.99"
 routerify = "1.1"
 hyper = "0.13"
-tokio-tungstenite = { version = "0.10", default-features = false, optional = true }
-headers = { version = "0.3", optional = true }
-futures = { version = "0.3", default-features = false, optional = true }
-tokio = { version = "0.2", features = ["rt-core"],  optional = true}
+headers = "0.3"
+tokio-tungstenite = { version = "0.10", default-features = false }
+futures = { version = "0.3", default-features = false }
+tokio = { version = "0.2", features = ["rt-core"] }
+
+serde = { version = "1.0", optional = true }
+serde_json = { version = "1.0", optional = true }
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "0.2", features = ["full"] }
+stream-body = "0.1"
+serde = { version = "1.0", features = ["derive"] }
+tokio-tungstenite = { version = "0.10", features = ["tls"] }

README.md

@@ -5,7 +5,7 @@
 
 # routerify-websocket
 
-An websocket extension for Routerify.
+The `websocket` support for the [Routerify](https://github.com/routerify/routerify) library.
 
 [Docs](https://docs.rs/routerify-websocket)
 
@@ -16,16 +16,75 @@ Add this to your `Cargo.toml` file:
 ```toml
 [dependencies]
 routerify = "1.1"
-routerify-websocket = "0.1.0"
+routerify-websocket = "1.0"
 ```
 
 ## Example
 
 ```rust
-use routerify_websocket;
+// Import `SinkExt` and `StreamExt` to send and read websocket messages.
+use futures::{SinkExt, StreamExt};
+use hyper::{Body, Response, Server};
+use routerify::{Router, RouterService};
+// Import websocket types.
+use routerify_websocket::{upgrade_ws, Message, WebSocket};
+use std::{convert::Infallible, net::SocketAddr};
 
-fn main() {
-  println!("{}", routerify_websocket::add(2, 3));
+// A handler for websocket connections.
+async fn ws_handler(ws: WebSocket) {
+    println!("New websocket connection: {}", ws.remote_addr());
+
+    // The `WebSocket` implements the `Sink` and `Stream` traits
+    // to read and write messages.
+    let (mut tx, mut rx) = ws.split();
+
+    // Read messages.
+    while let Some(msg) = rx.next().await {
+        let msg = msg.unwrap();
+
+        // Check message type and take appropriate actions.
+        if msg.is_text() {
+            println!("{}", msg.into_text().unwrap());
+        } else if msg.is_binary() {
+            println!("{:?}", msg.into_bytes());
+        }
+
+        // Send a text message.
+        let send_msg = Message::text("Hello world");
+        tx.send(send_msg).await.unwrap();
+    }
+}
+
+fn router() -> Router<Body, Infallible> {
+    // Create a router and specify the path and the handler for new websocket connections.
+    Router::builder()
+        // It will accept websocket connections at `/ws` path with any method type.
+        .any_method("/ws", upgrade_ws(ws_handler))
+        // It will accept http connections at `/` path.
+        .get("/", |_req| async move {
+            Ok(Response::new("I also serve http requests".into()))
+        })
+        .build()
+        .unwrap()
+}
+
+#[tokio::main]
+async fn main() {
+    let router = router();
+
+    // Create a Service from the router above to handle incoming requests.
+    let service = RouterService::new(router).unwrap();
+
+    // The address on which the server will be listening.
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3001));
+
+    // Create a server by passing the created service to `.serve` method.
+    let server = Server::bind(&addr).serve(service);
+
+    println!("App is running on: {}", addr);
+    if let Err(err) = server.await {
+        eprintln!("Server error: {}", err);
+    }
 }
 ```
 

examples/simple_example.rs

@@ -0,0 +1,63 @@
+// Import `SinkExt` and `StreamExt` to send and read websocket messages.
+use futures::{SinkExt, StreamExt};
+use hyper::{Body, Response, Server};
+use routerify::{Router, RouterService};
+// Import websocket types.
+use routerify_websocket::{upgrade_ws, Message, WebSocket};
+use std::{convert::Infallible, net::SocketAddr};
+
+// A handler for websocket connections.
+async fn ws_handler(ws: WebSocket) {
+    println!("New websocket connection: {}", ws.remote_addr());
+
+    // The `WebSocket` implements the `Sink` and `Stream` traits to read and write messages.
+    let (mut tx, mut rx) = ws.split();
+
+    // Read messages.
+    while let Some(msg) = rx.next().await {
+        let msg = msg.unwrap();
+
+        // Check message type and take appropriate actions.
+        if msg.is_text() {
+            println!("{}", msg.into_text().unwrap());
+        } else if msg.is_binary() {
+            println!("{:?}", msg.into_bytes());
+        }
+
+        // Send a text message.
+        let send_msg = Message::text("Hello world");
+        tx.send(send_msg).await.unwrap();
+    }
+}
+
+fn router() -> Router<Body, Infallible> {
+    // Create a router and specify the path and the handler for new websocket connections.
+    Router::builder()
+        // It will accept websocket connections at `/ws` path with any method type.
+        .any_method("/ws", upgrade_ws(ws_handler))
+        // It will accept http connections at `/` path.
+        .get("/", |_req| async move {
+            Ok(Response::new("I also serve http requests".into()))
+        })
+        .build()
+        .unwrap()
+}
+
+#[tokio::main]
+async fn main() {
+    let router = router();
+
+    // Create a Service from the router above to handle incoming requests.
+    let service = RouterService::new(router).unwrap();
+
+    // The address on which the server will be listening.
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3001));
+
+    // Create a server by passing the created service to `.serve` method.
+    let server = Server::bind(&addr).serve(service);
+
+    println!("App is running on: {}", addr);
+    if let Err(err) = server.await {
+        eprintln!("Server error: {}", err);
+    }
+}

examples/test.rs

@@ -1,21 +1,50 @@
+use futures::{Sink, SinkExt, StreamExt};
 use hyper::{Body, Request, Response, Server};
 use routerify::prelude::*;
 use routerify::{Middleware, Router, RouterService};
+use routerify_websocket::{upgrade_ws, upgrade_ws_with_config, Message, WebSocket, WebSocketConfig};
+use serde::{Deserialize, Serialize};
 use std::{convert::Infallible, net::SocketAddr};
+use tokio_tungstenite::{
+    tungstenite::protocol::{frame::coding::CloseCode, CloseFrame, Message as ClientMessage},
+    WebSocketStream,
+};
 
-async fn ws_handler(_: Request<Body>) -> Result<Response<Body>, Infallible> {
-    Ok(Response::new(Body::from("ws handler")))
+#[derive(Serialize, Deserialize, Debug)]
+struct User {
+    name: String,
+    roll: u64,
+}
+
+async fn ws_handler(ws: WebSocket) {
+    println!("new websocket connection: {}", ws.remote_addr());
+
+    let (mut tx, mut rx) = ws.split();
+
+    while let Some(msg) = rx.next().await {
+        let msg = msg.unwrap();
+
+        println!("{:?}", msg.close_reason());
+        println!("{}", String::from_utf8(msg.into_bytes()).unwrap());
+    }
 }
 
 async fn logger(req: Request<Body>) -> Result<Request<Body>, Infallible> {
     println!("{} {} {}", req.remote_addr(), req.method(), req.uri().path());
     Ok(req)
 }
 
+// A handler for "/about" page.
+async fn about_handler(_: Request<Body>) -> Result<Response<Body>, Infallible> {
+    println!("{:?}", std::thread::current().id());
+    Ok(Response::new(Body::from("About page")))
+}
+
 fn router() -> Router<Body, Infallible> {
     Router::builder()
         .middleware(Middleware::pre(logger))
-        .get("/ws/connect", ws_handler)
+        .get("/about", about_handler)
+        .any_method("/ws", upgrade_ws(ws_handler))
         .build()
         .unwrap()
 }
@@ -30,8 +59,25 @@ async fn main() {
 
     let server = Server::bind(&addr).serve(service);
 
+    tokio::spawn(async move {
+        tokio::time::delay_for(tokio::time::Duration::from_secs(3)).await;
+
+        let (mut ws, resp) = tokio_tungstenite::connect_async("ws://127.0.0.1:3001/ws")
+            .await
+            .unwrap();
+
+        println!("{:?}", resp.headers());
+
+        let msg = ClientMessage::text("hey");
+        ws.send(msg).await.unwrap();
+
+        ws.close(None).await.unwrap();
+
+        tokio::time::delay_for(tokio::time::Duration::from_secs(3)).await;
+    });
+
     println!("App is running on: {}", addr);
     if let Err(err) = server.await {
         eprintln!("Server error: {}", err);
     }
-}
+}

examples/test_stream_body.rs

@@ -0,0 +1,55 @@
+use futures::{Sink, SinkExt, StreamExt};
+use hyper::{Request, Response, Server};
+use routerify::{Router, RouterService};
+use routerify_websocket::{upgrade_ws, WebSocket};
+use std::{convert::Infallible, net::SocketAddr};
+use stream_body::StreamBody;
+use tokio::prelude::*;
+
+async fn ws_handler(ws: WebSocket) {
+    println!("new websocket connection: {}", ws.remote_addr());
+
+    let (mut tx, mut rx) = ws.split();
+
+    while let Some(msg) = rx.next().await {
+        let msg = msg.unwrap();
+
+        println!("{:?}", msg.close_reason());
+        println!("{}", String::from_utf8(msg.into_bytes()).unwrap());
+    }
+}
+
+fn router() -> Router<StreamBody, Infallible> {
+    Router::builder()
+        .any_method("/ws", upgrade_ws(ws_handler))
+        // Add options handler.
+        .options(
+            "/*",
+            |_req| async move { Ok(Response::new(StreamBody::from("Options"))) },
+        )
+        // Add 404 page handler.
+        .any(|_req| async move { Ok(Response::new(StreamBody::from("Not Found"))) })
+        // Add an error handler.
+        .err_handler(|err| async move { Response::new(StreamBody::from(format!("Error: {}", err))) })
+        .build()
+        .unwrap()
+}
+
+#[tokio::main]
+async fn main() {
+    let router = router();
+
+    // Create a Service from the router above to handle incoming requests.
+    let service = RouterService::new(router).unwrap();
+
+    // The address on which the server will be listening.
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3001));
+
+    // Create a server by passing the created service to `.serve` method.
+    let server = Server::bind(&addr).serve(service);
+
+    println!("App is running on: {}", addr);
+    if let Err(err) = server.await {
+        eprintln!("Server error: {}", err);
+    }
+}