hyper-function-core 0.1.1

Hyper Function
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use std::{collections::HashMap, convert::Infallible};

use hyper::{
    service::{make_service_fn, service_fn},
    Body, Request, Response, Server as HyperServer, StatusCode,
};
use rusty_ulid::generate_ulid_string;
use tokio::sync::mpsc;

use crate::{APP_ID, READ_CHAN_TX, SOCKET_CHANS};

use super::socket::{Action, Socket};

pub struct Server {
    pub addr: String,
}

impl Server {
    pub async fn handle_request(request: Request<Body>) -> Result<Response<Body>, Infallible> {
        if request.uri().path().eq("/hfn") {
            let bad_request = || {
                println!("bad request");
                return Ok(Response::builder()
                    .status(400)
                    .body(Body::from("Bad Request"))
                    .unwrap());
            };

            if !hyper_tungstenite::is_upgrade_request(&request) {
                return bad_request();
            }

            let qs = match request.uri().query() {
                Some(v) => v,
                None => return bad_request(),
            };

            let query: HashMap<String, String> = url::form_urlencoded::parse(&qs.as_bytes())
                .into_owned()
                .collect();

            let app_id = match query.get("aid") {
                Some(v) => v.to_string(),
                None => return bad_request(),
            };

            let client_id = match query.get("cid") {
                Some(v) => v.to_string(),
                None => return bad_request(),
            };

            let session_id = match query.get("sid") {
                Some(v) => v.to_string(),
                None => return bad_request(),
            };

            let client_version = match query.get("ver") {
                Some(v) => v.to_string(),
                None => return bad_request(),
            };

            let client_ts = match query.get("ts") {
                Some(v) => match v.parse::<u64>() {
                    Ok(v) => v,
                    Err(_) => return bad_request(),
                },
                None => return bad_request(),
            };

            if app_id.len() > 64
                || client_id.len() > 64
                || session_id.len() > 64
                || client_version.len() > 16
            {
                return bad_request();
            }

            if app_id.ne(unsafe { &APP_ID }) {
                return bad_request();
            }

            let (response, websocket) = match hyper_tungstenite::upgrade(request, None) {
                Ok(v) => v,
                Err(_) => return bad_request(),
            };

            tokio::spawn(async move {
                let stream = match websocket.await {
                    Ok(v) => v,
                    Err(_) => {
                        return ();
                    }
                };

                let socket = Socket {
                    id: generate_ulid_string(),
                    client_id,
                    session_id,
                    client_ts,
                    client_version,
                };

                let read_chan_tx = READ_CHAN_TX.get().unwrap().clone();
                let (socket_write_chan_tx, socket_write_chan_rx) =
                    mpsc::unbounded_channel::<Action>();

                let socket_id = socket.id.clone();
                let socket_chans = SOCKET_CHANS.get().unwrap();
                socket_chans.insert(socket_id.clone(), socket_write_chan_tx.clone());

                socket
                    .accept_ws(
                        stream,
                        read_chan_tx,
                        socket_write_chan_tx,
                        socket_write_chan_rx,
                    )
                    .await;

                // clean up
                socket_chans.remove(&socket_id);
            });

            // Return the response so the spawned future can continue.
            Ok(response)
        } else {
            let mut response = Response::new(Body::empty());
            *response.status_mut() = StatusCode::NOT_FOUND;

            Ok(response)
        }
    }
    pub async fn listen(&self) {
        let addr: std::net::SocketAddr = self.addr.parse().expect("fail to parse addr");

        let server = HyperServer::bind(&addr).serve(make_service_fn(|_| async {
            Ok::<_, Infallible>(service_fn(Server::handle_request))
        }));

        println!("Listening on {}", self.addr);
        if let Err(e) = server.await {
            eprintln!("server error: {}", e);
        }
    }
}