rate-core 0.41.0

The RillRate embedded HTTP server.
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

use super::{Group, Router};
use crate::actors::client_session::ClientSession;
use crate::actors::supervisor::Supervisor;
//use crate::info::TRACERS;
use anyhow::Error;
use async_trait::async_trait;
use meio::{ActionHandler, Context, Eliminated, IdOf, InteractionHandler};
use meio_connect::headers::HeaderValue;
use meio_connect::hyper::{header, Body, Response, StatusCode};
use meio_connect::server::{DirectPath, NoParameters, Req, WebRoute, WsReq, WsRoute};
use rill_protocol::io::client::ClientProtocol;

impl<T: Supervisor> Router<T> {
    pub(super) async fn init_external(&mut self, ctx: &mut Context<Self>) -> Result<(), Error> {
        // TODO: Make it configurable
        let route = WebRoute::new(IndexToUi, ctx.address().clone());
        self.external_server.add_route(route).await?;

        let route = WsRoute::new(ClientLive, ctx.address().clone());
        self.external_server.add_route(route).await?;

        Ok(())
    }
}

struct IndexToUi;

impl DirectPath for IndexToUi {
    type Output = NoParameters;
    type Parameter = ();
    fn paths() -> &'static [&'static str] {
        &["/", "/index.html"]
    }
}

#[async_trait]
impl<T: Supervisor> InteractionHandler<Req<IndexToUi>> for Router<T> {
    async fn handle(
        &mut self,
        _: Req<IndexToUi>,
        _ctx: &mut Context<Self>,
    ) -> Result<Response<Body>, Error> {
        let mut response = Response::builder()
            .status(StatusCode::TEMPORARY_REDIRECT)
            .body(Body::empty())?;
        // My eyes cry
        response
            .headers_mut()
            .insert(header::LOCATION, HeaderValue::from_static("/ui/"));
        Ok(response)
    }
}

struct ClientLive;

impl DirectPath for ClientLive {
    type Output = NoParameters;
    type Parameter = ClientProtocol;

    fn paths() -> &'static [&'static str] {
        &["/live/client"]
    }
}

#[async_trait]
impl<T: Supervisor> ActionHandler<WsReq<ClientLive>> for Router<T> {
    async fn handle(
        &mut self,
        req: WsReq<ClientLive>,
        ctx: &mut Context<Self>,
    ) -> Result<(), Error> {
        if !ctx.is_terminating() {
            //let ip = req.stream.addr().ip();
            if self.active_clients.has_slot() {
                let session_actor = ClientSession::new(
                    self.supervisor.clone(),
                    req.stream,
                    self.registry.clone(),
                    self.global_acl.clone(),
                );
                let addr = ctx.spawn_actor(session_actor, Group::Externals);
                self.active_clients.acquire(addr);
            } else {
                /* TODO: Good idea, but use a special server-level tracer
                let alert = format!(
                    "Active clients {} limit reached.",
                    self.active_clients.limit().total
                );
                TRACERS.alerts.alert(alert);
                */
                log::warn!("Limit of active clients reached.");
            }
        } else {
            log::warn!("Incoming client connection rejected, because the server is terminating.");
        }
        Ok(())
    }
}

#[async_trait]
impl<T: Supervisor> Eliminated<ClientSession<T>> for Router<T> {
    async fn handle(
        &mut self,
        id: IdOf<ClientSession<T>>,
        _ctx: &mut Context<Self>,
    ) -> Result<(), Error> {
        self.active_clients.release(id);
        Ok(())
    }
}