plane 0.5.5

Session backend orchestrator for ambitious browser-based apps.
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

use crate::database::{connect::ConnectError, PlaneDatabase};
use chrono::{DateTime, Utc};
use plane_common::{
    names::{AnyNodeName, ControllerName},
    typed_socket::Handshake,
    types::{ClusterName, ConnectRequest, ConnectResponse, NodeId},
    PlaneClient,
};
use std::net::IpAddr;
use url::Url;

#[derive(Clone)]
pub struct Controller {
    pub db: PlaneDatabase,
    pub id: ControllerName,
    pub client: PlaneClient,
    pub default_cluster: Option<ClusterName>,
}

pub struct NodeHandle {
    pub id: NodeId,
    pub controller: ControllerName,
    db: Option<PlaneDatabase>,
    connection_start_time: DateTime<Utc>,
}

impl Drop for NodeHandle {
    fn drop(&mut self) {
        let db = self
            .db
            .take()
            .expect("self.db is always Some before dropped.");
        let id = self.id;
        let controller = self.controller.clone();
        let connection_start_time = self.connection_start_time;
        tokio::spawn(async move {
            if let Err(err) = db
                .node()
                .mark_offline(id, &controller, connection_start_time)
                .await
            {
                tracing::error!(?err, "Failed to mark node offline.");
            }
        });
    }
}

impl Controller {
    pub async fn register_node(
        &self,
        handshake: Handshake,
        cluster: Option<&ClusterName>,
        ip: IpAddr,
    ) -> Result<NodeHandle, sqlx::Error> {
        let name = AnyNodeName::try_from(handshake.name).map_err(|_| {
            sqlx::Error::Decode("Failed to decode node name from handshake.".into())
        })?;
        let kind = name.kind();
        let plane_version = handshake.version;

        let (node_id, connection_start_time) = self
            .db
            .node()
            .register(cluster, &name, kind, &self.id, &plane_version, ip)
            .await?;

        Ok(NodeHandle {
            id: node_id,
            db: Some(self.db.clone()),
            controller: self.id.clone(),
            connection_start_time,
        })
    }

    pub async fn new(
        db: PlaneDatabase,
        id: ControllerName,
        controller_url: Url,
        default_cluster: Option<ClusterName>,
    ) -> Self {
        let client = PlaneClient::new(controller_url);

        Self {
            db,
            id,
            client,
            default_cluster,
        }
    }

    pub async fn connect(
        &self,
        connect_request: &ConnectRequest,
    ) -> Result<ConnectResponse, ConnectError> {
        let response = self
            .db
            .connect(self.default_cluster.as_ref(), connect_request, &self.client)
            .await?;

        Ok(response)
    }
}