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

use super::link;
use crate::actors::router::Router;
use crate::config::ServerConfig;
use anyhow::Error;
use async_trait::async_trait;
use meio::{Actor, Context, Eliminated, IdOf, InteractionHandler, InterruptedBy, StartedBy};
use meio_connect::server::{HttpServer, HttpServerLink};
use rill_client::actors::broadcaster::Broadcaster;

/// Embedded node.
pub struct RillServer {
    server_config: ServerConfig,
    public_server: Option<HttpServerLink>,
    private_server: Option<HttpServerLink>,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Group {
    Tuning,
    Broadcaster,
    HttpServer,
    Endpoints,
}

impl Actor for RillServer {
    type GroupBy = Group;
}

impl RillServer {
    /// Create a new instance of an embedded node.
    pub fn new(server_config: Option<ServerConfig>) -> Self {
        Self {
            server_config: server_config.unwrap_or_default(),
            public_server: None,
            private_server: None,
        }
    }
}

#[async_trait]
impl<T: Actor> StartedBy<T> for RillServer {
    async fn handle(&mut self, ctx: &mut Context<Self>) -> Result<(), Error> {
        ctx.termination_sequence(vec![
            Group::Tuning,
            Group::Broadcaster,
            Group::HttpServer,
            Group::Endpoints,
        ]);

        // Starting all basic actors
        let extern_addr = format!("{}:{}", self.server_config.server_address(), 9090).parse()?;
        let extern_http_server_actor = HttpServer::new(extern_addr);
        let extern_http_server = ctx.spawn_actor(extern_http_server_actor, Group::HttpServer);
        self.public_server = Some(extern_http_server.link());

        let inner_addr = format!("127.0.0.1:{}", 0).parse()?;
        let inner_http_server_actor = HttpServer::new(inner_addr);
        let inner_http_server = ctx.spawn_actor(inner_http_server_actor, Group::HttpServer);
        self.private_server = Some(inner_http_server.link());

        let exporter_actor = Broadcaster::new();
        let exporter = ctx.spawn_actor(exporter_actor, Group::Broadcaster);

        let server_actor = Router::new(
            inner_http_server.link(),
            extern_http_server.link(),
            exporter.link(),
        );
        let _router = ctx.spawn_actor(server_actor, Group::Endpoints);

        Ok(())
    }
}

#[async_trait]
impl<T: Actor> InterruptedBy<T> for RillServer {
    async fn handle(&mut self, ctx: &mut Context<Self>) -> Result<(), Error> {
        ctx.shutdown();
        Ok(())
    }
}

#[async_trait]
impl Eliminated<Broadcaster> for RillServer {
    async fn handle(
        &mut self,
        _id: IdOf<Broadcaster>,
        _ctx: &mut Context<Self>,
    ) -> Result<(), Error> {
        log::info!("Broadcaster finished");
        Ok(())
    }
}

#[async_trait]
impl Eliminated<HttpServer> for RillServer {
    async fn handle(
        &mut self,
        _id: IdOf<HttpServer>,
        _ctx: &mut Context<Self>,
    ) -> Result<(), Error> {
        log::info!("HttpServer finished");
        Ok(())
    }
}

#[async_trait]
impl Eliminated<Router> for RillServer {
    async fn handle(&mut self, _id: IdOf<Router>, _ctx: &mut Context<Self>) -> Result<(), Error> {
        log::info!("Router finished");
        Ok(())
    }
}

#[async_trait]
impl InteractionHandler<link::WaitPublicEndpoint> for RillServer {
    async fn handle(
        &mut self,
        _msg: link::WaitPublicEndpoint,
        _ctx: &mut Context<Self>,
    ) -> Result<HttpServerLink, Error> {
        // `public_server` always available here since it's attached in `StartedBy` handler
        self.public_server
            .clone()
            .ok_or_else(|| Error::msg("no public server"))
    }
}

#[async_trait]
impl InteractionHandler<link::WaitPrivateEndpoint> for RillServer {
    async fn handle(
        &mut self,
        _msg: link::WaitPrivateEndpoint,
        _ctx: &mut Context<Self>,
    ) -> Result<HttpServerLink, Error> {
        // `private_server` always available here since it's attached in `StartedBy` handler
        self.private_server
            .clone()
            .ok_or_else(|| Error::msg("no private server"))
    }
}