add async feature

Author: stevefan1999-personal

Cargo.toml

@@ -11,20 +11,70 @@ categories = ["embedded", "no-std", "network-programming"]
 readme = "README.md"
 
 [dependencies]
-sha1 = "0.10.1"
-heapless = "0.7.14"
-byteorder = { version = "1.4.3", default-features = false }
-httparse = { version = "1.7.1", default-features = false }
-rand_core = "0.6.3"
+anyhow = "1.0.58"
+async-std = { version = "1.12.0", optional = true }
 base64 = { version = "0.13.0", default-features = false }
 base64-simd = { version = "0.5.0", default-features = false, optional = true }
+byteorder = { version = "1.4.3", default-features = false }
 cfg-if = "1.0.0"
+core2 = "0.4.0"
+futures = { version = "0.3.21", optional = true }
+heapless = "0.7.14"
+httparse = { version = "1.7.1", default-features = false }
+rand_core = "0.6.3"
+sha1 = "0.10.1"
+smol = { version = "1.2.5", optional = true }
+tokio = { version = "1.19.2", features = ["io-util"], optional = true }
 
 [dev-dependencies]
+async-std = { version = "1.12.0", features = ["attributes"] }
+async-trait = "0.1.56"
+cfg-if = "1.0.0"
+once_cell = "1.12.0"
 rand = "0.8.5"
+route-recognizer = "0.3.1"
+smol = "1.2.5"
+smol-potat = { version = "1.1.2", features = ["auto"] }
+tokio = { version = "1.19.2", features = ["macros", "net", "rt-multi-thread", "io-util"] }
+tokio-stream = { version = "0.1.9", features = ["net"] }
 
 # see readme for no_std support
 [features]
 default = ["std"]
 # default = []
 std = []
+async = ["std"]
+tokio = ["dep:tokio", "async"]
+futures = ["dep:futures", "async"]
+smol = ["dep:smol", "async"]
+async-std = ["dep:async-std", "async"]
+
+[[example]]
+name = "server_tokio"
+path = "examples/server_async/main.rs"
+required-features = ["tokio"]
+
+[[example]]
+name = "server_smol"
+path = "examples/server_async/main.rs"
+required-features = ["smol"]
+
+[[example]]
+name = "server_async_std"
+path = "examples/server_async/main.rs"
+required-features = ["async-std"]
+
+[[example]]
+name = "client_tokio"
+path = "examples/client_async/main.rs"
+required-features = ["tokio"]
+
+[[example]]
+name = "client_smol"
+path = "examples/client_async/main.rs"
+required-features = ["smol"]
+
+[[example]]
+name = "client_async_std"
+path = "examples/client_async/main.rs"
+required-features = ["async-std"]

examples/client.rs renamed to examples/client/main.rs

@@ -13,13 +13,15 @@ use embedded_websocket::{
     framer::{Framer, FramerError, ReadResult},
     WebSocketClient, WebSocketCloseStatusCode, WebSocketOptions, WebSocketSendMessageType,
 };
-use std::{error::Error, net::TcpStream};
+use std::net::TcpStream;
 
-fn main() -> Result<(), FramerError<impl Error>> {
+fn main() -> Result<(), FramerError> {
     // open a TCP stream to localhost port 1337
     let address = "127.0.0.1:1337";
     println!("Connecting to: {}", address);
-    let mut stream = TcpStream::connect(address).map_err(FramerError::Io)?;
+    let mut stream = TcpStream::connect(address)
+        .map_err(anyhow::Error::new)
+        .map_err(FramerError::Io)?;
     println!("Connected.");
 
     let mut read_buf = [0; 4000];

examples/client_async/main.rs

@@ -0,0 +1,87 @@
+// The MIT License (MIT)
+// Copyright (c) 2019 David Haig
+
+// Demo websocket client connecting to localhost port 1337.
+// This will initiate a websocket connection to path /chat. The demo sends a simple "Hello, World!"
+// message and expects an echo of the same message as a reply.
+// It will then initiate a close handshake, wait for a close response from the server,
+// and terminate the connection.
+// Note that we are using the standard library in the demo and making use of the framer helper module
+// but the websocket library remains no_std (see client_full for an example without the framer helper module)
+
+use embedded_websocket::{
+    framer::{Framer, FramerError, ReadResult},
+    WebSocketClient, WebSocketCloseStatusCode, WebSocketOptions, WebSocketSendMessageType,
+};
+
+cfg_if::cfg_if! {
+    if #[cfg(feature = "tokio")] {
+        use tokio::net::TcpStream;
+    } else if #[cfg(feature = "smol")] {
+        use smol::net::TcpStream;
+    } else if #[cfg(feature = "async-std")] {
+        use async_std::net::TcpStream;
+    }
+}
+
+#[cfg_attr(feature = "async-std", async_std::main)]
+#[cfg_attr(feature = "tokio", tokio::main)]
+#[cfg_attr(feature = "smol", smol_potat::main)]
+async fn main() -> Result<(), FramerError> {
+    // open a TCP stream to localhost port 1337
+    let address = "127.0.0.1:1337";
+    println!("Connecting to: {}", address);
+    let mut stream = TcpStream::connect(address)
+        .await
+        .map_err(anyhow::Error::new)
+        .map_err(FramerError::Io)?;
+    println!("Connected.");
+
+    let mut read_buf = [0; 4000];
+    let mut read_cursor = 0;
+    let mut write_buf = [0; 4000];
+    let mut frame_buf = [0; 4000];
+    let mut websocket = WebSocketClient::new_client(rand::thread_rng());
+
+    // initiate a websocket opening handshake
+    let websocket_options = WebSocketOptions {
+        path: "/chat",
+        host: "localhost",
+        origin: "http://localhost:1337",
+        sub_protocols: None,
+        additional_headers: None,
+    };
+
+    let mut framer = Framer::new(
+        &mut read_buf,
+        &mut read_cursor,
+        &mut write_buf,
+        &mut websocket,
+    );
+    framer
+        .connect_async(&mut stream, &websocket_options)
+        .await?;
+
+    let message = "Hello, World!";
+    framer
+        .write_async(
+            &mut stream,
+            WebSocketSendMessageType::Text,
+            true,
+            message.as_bytes(),
+        )
+        .await?;
+
+    while let ReadResult::Text(s) = framer.read_async(&mut stream, &mut frame_buf).await? {
+        println!("Received: {}", s);
+
+        // close the websocket after receiving the first reply
+        framer
+            .close_async(&mut stream, WebSocketCloseStatusCode::NormalClosure, None)
+            .await?;
+        println!("Sent close handshake");
+    }
+
+    println!("Connection closed");
+    Ok(())
+}

examples/server.rs renamed to examples/server/main.rs

@@ -3,32 +3,33 @@
 
 // Demo websocket server that listens on localhost port 1337.
 // If accessed from a browser it will return a web page that will automatically attempt to
-// open a websocket connection to itself. Alternatively, the client.rs example can be used to
+// open a websocket connection to itself. Alternatively, the main example can be used to
 // open a websocket connection directly. The server will echo all Text and Ping messages back to
 // the client as well as responding to any opening and closing handshakes.
 // Note that we are using the standard library in the demo but the websocket library remains no_std
 
 use embedded_websocket as ws;
+use httparse::Request;
+use once_cell::sync::Lazy;
+use route_recognizer::Router;
+use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream};
 use std::str::Utf8Error;
 use std::thread;
-use std::{
-    io::{Read, Write},
-    usize,
-};
 use ws::framer::ReadResult;
 use ws::{
     framer::{Framer, FramerError},
-    WebSocketContext, WebSocketSendMessageType, WebSocketServer,
+    WebSocketSendMessageType, WebSocketServer,
 };
 
 type Result<T> = std::result::Result<T, WebServerError>;
 
 #[derive(Debug)]
 pub enum WebServerError {
     Io(std::io::Error),
-    Framer(FramerError<std::io::Error>),
+    Framer(FramerError),
     WebSocket(ws::Error),
+    HttpError(String),
     Utf8Error,
 }
 
@@ -38,8 +39,8 @@ impl From<std::io::Error> for WebServerError {
     }
 }
 
-impl From<FramerError<std::io::Error>> for WebServerError {
-    fn from(err: FramerError<std::io::Error>) -> WebServerError {
+impl From<FramerError> for WebServerError {
+    fn from(err: FramerError) -> WebServerError {
         WebServerError::Framer(err)
     }
 }
@@ -77,13 +78,24 @@ fn main() -> std::io::Result<()> {
     Ok(())
 }
 
-fn handle_client(mut stream: TcpStream) -> Result<()> {
-    println!("Client connected {}", stream.peer_addr()?);
-    let mut read_buf = [0; 4000];
-    let mut read_cursor = 0;
+type Handler = Box<dyn Fn(&mut TcpStream, &Request) -> Result<()> + Send + Sync>;
+
+static ROUTER: Lazy<Router<Handler>> = Lazy::new(|| {
+    let mut router = Router::new();
+    router.add("/chat", Box::new(handle_chat) as Handler);
+    router.add("/", Box::new(handle_root) as Handler);
+    router
+});
 
-    if let Some(websocket_context) = read_header(&mut stream, &mut read_buf, &mut read_cursor)? {
+fn handle_chat(stream: &mut TcpStream, req: &Request) -> Result<()> {
+    println!("Received chat request: {:?}", req.path);
+
+    if let Some(websocket_context) =
+        ws::read_http_header(req.headers.iter().map(|f| (f.name, f.value)))?
+    {
         // this is a websocket upgrade HTTP request
+        let mut read_buf = [0; 4000];
+        let mut read_cursor = 0;
         let mut write_buf = [0; 4000];
         let mut frame_buf = [0; 4000];
         let mut websocket = WebSocketServer::new_server();
@@ -95,85 +107,65 @@ fn handle_client(mut stream: TcpStream) -> Result<()> {
         );
 
         // complete the opening handshake with the client
-        framer.accept(&mut stream, &websocket_context)?;
+        framer.accept(stream, &websocket_context)?;
         println!("Websocket connection opened");
 
         // read websocket frames
-        while let ReadResult::Text(text) = framer.read(&mut stream, &mut frame_buf)? {
+        while let ReadResult::Text(text) = framer.read(stream, &mut frame_buf)? {
             println!("Received: {}", text);
 
             // send the text back to the client
             framer.write(
-                &mut stream,
+                stream,
                 WebSocketSendMessageType::Text,
                 true,
-                text.as_bytes(),
+                format!("shut the fuck up {}", text).as_bytes(),
             )?
         }
 
         println!("Closing websocket connection");
-        Ok(())
-    } else {
-        Ok(())
     }
+
+    Ok(())
 }
 
-fn read_header(
-    stream: &mut TcpStream,
-    read_buf: &mut [u8],
-    read_cursor: &mut usize,
-) -> Result<Option<WebSocketContext>> {
-    loop {
-        let mut headers = [httparse::EMPTY_HEADER; 16];
-        let mut request = httparse::Request::new(&mut headers);
-
-        let received_size = stream.read(&mut read_buf[*read_cursor..])?;
-
-        match request
-            .parse(&read_buf[..*read_cursor + received_size])
-            .unwrap()
-        {
-            httparse::Status::Complete(len) => {
-                // if we read exactly the right amount of bytes for the HTTP header then read_cursor would be 0
-                *read_cursor += received_size - len;
-                let headers = request.headers.iter().map(|f| (f.name, f.value));
-                match ws::read_http_header(headers)? {
-                    Some(websocket_context) => match request.path {
-                        Some("/chat") => {
-                            return Ok(Some(websocket_context));
-                        }
-                        _ => return_404_not_found(stream, request.path)?,
-                    },
-                    None => {
-                        handle_non_websocket_http_request(stream, request.path)?;
-                    }
-                }
-                return Ok(None);
-            }
-            // keep reading while the HTTP header is incomplete
-            httparse::Status::Partial => *read_cursor += received_size,
-        }
-    }
+fn handle_root(stream: &mut TcpStream, _req: &Request) -> Result<()> {
+    stream.write_all(&ROOT_HTML.as_bytes())?;
+    Ok(())
 }
 
-fn handle_non_websocket_http_request(stream: &mut TcpStream, path: Option<&str>) -> Result<()> {
-    println!("Received file request: {:?}", path);
+fn handle_client(mut stream: TcpStream) -> Result<()> {
+    println!("Client connected {}", stream.peer_addr()?);
+    let mut read_buf = [0; 4000];
+    let mut read_cursor = 0;
 
-    match path {
-        Some("/") => stream.write_all(&ROOT_HTML.as_bytes())?,
-        unknown_path => {
-            return_404_not_found(stream, unknown_path)?;
+    let mut headers = vec![httparse::EMPTY_HEADER; 8];
+    let received_size = stream.read(&mut read_buf[read_cursor..])?;
+    let request = loop {
+        let mut request = Request::new(&mut headers);
+        match request.parse(&read_buf[..read_cursor + received_size]) {
+            Ok(httparse::Status::Partial) => read_cursor += received_size,
+            Ok(httparse::Status::Complete(_)) => break request,
+            Err(httparse::Error::TooManyHeaders) => {
+                headers.resize(headers.len() * 2, httparse::EMPTY_HEADER)
+            }
+            _ => panic!("http parser error"),
         }
     };
 
-    Ok(())
-}
-
-fn return_404_not_found(stream: &mut TcpStream, unknown_path: Option<&str>) -> Result<()> {
-    println!("Unknown path: {:?}", unknown_path);
-    let html = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
-    stream.write_all(&html.as_bytes())?;
-    Ok(())
+    match ROUTER.recognize(request.path.unwrap_or("/")) {
+        Ok(handler) => handler.handler()(&mut stream, &request),
+        Err(e) => {
+            println!("Unknown path: {:?}", request.path);
+            let html = format!(
+                "HTTP/1.1 404 Not Found\r\nContent-Length: {len}\r\nConnection: close\r\n\r\n{msg}",
+                len = e.len(),
+                msg = e
+            );
+            stream.write_all(&html.as_bytes())?;
+            Err(WebServerError::HttpError(e))
+        }
+    }
 }
 
 const ROOT_HTML : &str = "HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=UTF-8\r\nContent-Length: 2590\r\nConnection: close\r\n\r\n<!doctype html>