oxidized 0.1.0

A hyper-performant, modular, and asynchronous web framework for Rust, built on hyper and tokio. Featuring a Tower-inspired, Service-based architecture for maximum performance and composability.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

use crate::{
    http_request::Request,
    response::Response,
    ws::{CloseFrame, Message, WebSocket},
    Result,
};
use base64::{engine::general_purpose, Engine as _};
use futures_util::Future;
use hyper::{
    self,
    header::{CONNECTION, SEC_WEBSOCKET_ACCEPT, SEC_WEBSOCKET_KEY, UPGRADE},
    StatusCode,
};
use hyper_util::rt::TokioIo;
use sha1::{Digest, Sha1};
use tokio_tungstenite::{
    tungstenite::{self, protocol::Role},
    WebSocketStream,
};

pub fn upgrade<F, Fut>(f: F) -> impl Fn(Request) -> Result<Response>
where
    F: Fn(WebSocket) -> Fut + Clone + Send + Sync + 'static,
    Fut: Future<Output = ()> + Send + 'static,
{
    move |mut req: Request| {
        let key = req.inner().headers().get(SEC_WEBSOCKET_KEY).cloned();

        let f = f.clone();
        tokio::spawn(async move {
            match hyper::upgrade::on(req.inner_mut()).await {
                Ok(upgraded) => {
                    let io = TokioIo::new(upgraded);
                    let stream = WebSocketStream::from_raw_socket(io, Role::Server, None).await;
                    let ws = WebSocket::new(stream);
                    f(ws).await;
                }
                Err(e) => {
                    eprintln!("websocket upgrade error: {}", e);
                }
            }
        });

        let mut res = Response::new("");
        *res.inner_mut().status_mut() = StatusCode::SWITCHING_PROTOCOLS;

        if let Some(key) = key {
            let mut hasher = Sha1::new();
            hasher.update(key.as_bytes());
            hasher.update(b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"); // WebSocket magic string
            let accept_key = general_purpose::STANDARD.encode(hasher.finalize());
            res.inner_mut()
                .headers_mut()
                .insert(SEC_WEBSOCKET_ACCEPT, accept_key.parse().unwrap());
        }

        res.inner_mut()
            .headers_mut()
            .insert(CONNECTION, "upgrade".parse().unwrap());
        res.inner_mut()
            .headers_mut()
            .insert(UPGRADE, "websocket".parse().unwrap());

        Ok(res)
    }
}

impl From<tungstenite::Message> for Message {
    fn from(msg: tungstenite::Message) -> Self {
        match msg {
            tungstenite::Message::Text(text) => Message::Text(text),
            tungstenite::Message::Binary(data) => Message::Binary(data),
            tungstenite::Message::Ping(data) => Message::Ping(data),
            tungstenite::Message::Pong(data) => Message::Pong(data),
            tungstenite::Message::Close(Some(frame)) => Message::Close(Some(CloseFrame {
                code: frame.code.into(),
                reason: frame.reason.into(),
            })),
            tungstenite::Message::Close(None) => Message::Close(None),
            tungstenite::Message::Frame(_) => unimplemented!(),
        }
    }
}