new-home-proxy 0.1.2

This is a part of the New Home IoT System. It is used to make the core available in the www.
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

use std::sync::Arc;
use std::time::SystemTime;

use actix::clock::Duration;
use actix::{Actor, AsyncContext, Handler, StreamHandler};
use actix_web_actors::ws::{Message as WebsocketMessage, ProtocolError, WebsocketContext};
use serde_json::to_string as json_to_string;

use crate::communication::{
    ProxyRequest, ProxyResponse, WebsocketCommand, WebsocketSignal, WrappedMessage,
};
use crate::server::client_manager::ClientManager;

pub struct WebsocketHandler {
    last_ping: SystemTime,
    client_manager: Arc<ClientManager>,
    client_id: String,
}

impl WebsocketHandler {
    pub fn new(client_manager: Arc<ClientManager>, client_id: String) -> Self {
        WebsocketHandler {
            last_ping: SystemTime::now(),
            client_manager,
            client_id,
        }
    }
}

impl Actor for WebsocketHandler {
    type Context = WebsocketContext<Self>;
}

impl StreamHandler<Result<WebsocketMessage, ProtocolError>> for WebsocketHandler {
    fn handle(
        &mut self,
        message: Result<WebsocketMessage, ProtocolError>,
        ctx: &mut Self::Context,
    ) {
        match message {
            Ok(WebsocketMessage::Text(message)) => {
                if let Ok(message) = serde_json::from_str::<WrappedMessage<ProxyResponse>>(&message)
                {
                    if let Err(error) = self
                        .client_manager
                        .response(message.id.clone(), message.into_inner())
                    {
                        println!("Could not respond to message: {}", error);
                    }
                }
            }
            Ok(WebsocketMessage::Ping(_)) => {
                self.last_ping = SystemTime::now();
                ctx.pong(&[]);
            }
            _ => {}
        }
    }

    fn started(&mut self, ctx: &mut Self::Context) {
        self.last_ping = SystemTime::now();

        if let Err(error) = self
            .client_manager
            .register_client(self.client_id.clone(), ctx.address())
        {
            println!("Client could not be registered: {}", error);

            ctx.close(None);
        }
    }

    fn finished(&mut self, _: &mut Self::Context) {
        if let Err(_) = self.client_manager.unregister_client(&self.client_id) {}
    }
}

impl Handler<WrappedMessage<ProxyRequest>> for WebsocketHandler {
    type Result = ();

    fn handle(
        &mut self,
        msg: WrappedMessage<ProxyRequest>,
        ctx: &mut Self::Context,
    ) -> Self::Result {
        ctx.text(json_to_string(&msg).unwrap_or_default());

        ()
    }
}

impl Handler<WebsocketCommand> for WebsocketHandler {
    type Result = ();

    fn handle(&mut self, command: WebsocketCommand, ctx: &mut Self::Context) -> Self::Result {
        match command.0 {
            WebsocketSignal::Close => {
                ctx.close(None);

                self.client_manager
                    .unregister_client(&self.client_id)
                    .unwrap_or_default();
            }
            WebsocketSignal::CheckPing => {
                let ping_age = SystemTime::now()
                    .duration_since(self.last_ping)
                    .unwrap_or(Duration::from_secs(99))
                    .as_millis();

                if ping_age > 1500 {
                    ctx.close(None);

                    self.client_manager
                        .unregister_client(&self.client_id)
                        .unwrap_or_default();
                }
            }
            _ => {}
        }
    }
}