kaspa-grpc-server 0.0.4

Kaspa gRPC 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

use crate::{connection_handler::ConnectionHandler, manager::Manager};
use kaspa_core::debug;
use kaspa_notify::notifier::Notifier;
use kaspa_rpc_core::{api::rpc::DynRpcService, notify::connection::ChannelConnection, Notification, RpcResult};
use kaspa_utils::networking::NetAddress;
use std::{ops::Deref, sync::Arc};
use tokio::sync::{mpsc::channel as mpsc_channel, oneshot::Sender as OneshotSender};

pub struct Adaptor {
    /// If a server was started, it will get cleaned up when this sender is dropped or invoked
    _server_termination: Option<OneshotSender<()>>,

    /// An object for handling new connections coming from clients
    connection_handler: ConnectionHandler,

    /// An object for managing a list of active connections
    manager: Manager,

    /// The network address of the server
    serve_address: NetAddress,
}

impl Adaptor {
    fn new(
        server_termination: Option<OneshotSender<()>>,
        connection_handler: ConnectionHandler,
        manager: Manager,
        serve_address: NetAddress,
    ) -> Self {
        Self { _server_termination: server_termination, connection_handler, manager, serve_address }
    }

    pub fn server(
        serve_address: NetAddress,
        manager: Manager,
        core_service: DynRpcService,
        core_notifier: Arc<Notifier<Notification, ChannelConnection>>,
    ) -> Arc<Self> {
        let (manager_sender, manager_receiver) = mpsc_channel(Self::manager_channel_size());
        let connection_handler = ConnectionHandler::new(manager_sender, core_service.clone(), core_notifier);
        let server_termination = connection_handler.serve(serve_address);
        let adaptor = Arc::new(Adaptor::new(Some(server_termination), connection_handler, manager, serve_address));
        adaptor.manager.clone().start_event_loop(manager_receiver);
        adaptor.start();
        adaptor
    }

    pub fn serve_address(&self) -> NetAddress {
        self.serve_address
    }

    pub fn start(&self) {
        self.connection_handler.start()
    }

    /// Terminates all connections and cleans up any additional async resources
    pub async fn stop(&self) -> RpcResult<()> {
        debug!("GRPC, Stopping the adaptor");
        self.terminate_all_connections();
        self.connection_handler.stop().await?;
        Ok(())
    }

    pub fn manager_channel_size() -> usize {
        128
    }
}

/// Expose all public `Manager` methods directly through the `Adaptor`
impl Deref for Adaptor {
    type Target = Manager;

    fn deref(&self) -> &Self::Target {
        &self.manager
    }
}