kaspa-rpc-core 0.15.0

Kaspa RPC core
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

//!
//! Client-side RPC helper for handling connection and disconnection events.
//!

use crate::error::RpcResult;
use std::sync::{Arc, Mutex};
use workflow_core::channel::Multiplexer;

/// RPC channel control operations
#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
pub enum RpcState {
    /// RpcApi channel open (connected)
    Connected,
    /// RpcApi channel close (disconnected)
    #[default]
    Disconnected,
}

#[derive(Default)]
struct Inner {
    // Current channel state
    state: Mutex<RpcState>,
    // MPMC channel for [`RpcCtlOp`] operations.
    multiplexer: Multiplexer<RpcState>,
    // Optional Connection descriptor such as a connection URL.
    descriptor: Mutex<Option<String>>,
}

/// RPC channel control helper. This is a companion
/// struct to [`RpcApi`](crate::api::rpc::RpcApi) that
/// provides signaling for RPC open/close events as
/// well as an optional connection descriptor (URL).
#[derive(Default, Clone)]
pub struct RpcCtl {
    inner: Arc<Inner>,
}

impl RpcCtl {
    pub fn new() -> Self {
        Self { inner: Arc::new(Inner::default()) }
    }

    pub fn with_descriptor<Str: ToString>(descriptor: Option<Str>) -> Self {
        if let Some(descriptor) = descriptor {
            Self { inner: Arc::new(Inner { descriptor: Mutex::new(Some(descriptor.to_string())), ..Inner::default() }) }
        } else {
            Self::default()
        }
    }

    /// Obtain internal multiplexer (MPMC channel for [`RpcState`] operations)
    pub fn multiplexer(&self) -> &Multiplexer<RpcState> {
        &self.inner.multiplexer
    }

    pub fn is_connected(&self) -> bool {
        *self.inner.state.lock().unwrap() == RpcState::Connected
    }

    pub fn state(&self) -> RpcState {
        *self.inner.state.lock().unwrap()
    }

    /// Signal open to all listeners (async)
    pub async fn signal_open(&self) -> RpcResult<()> {
        *self.inner.state.lock().unwrap() = RpcState::Connected;
        Ok(self.inner.multiplexer.broadcast(RpcState::Connected).await?)
    }

    /// Signal close to all listeners (async)
    pub async fn signal_close(&self) -> RpcResult<()> {
        *self.inner.state.lock().unwrap() = RpcState::Disconnected;
        Ok(self.inner.multiplexer.broadcast(RpcState::Disconnected).await?)
    }

    /// Try signal open to all listeners (sync)
    pub fn try_signal_open(&self) -> RpcResult<()> {
        *self.inner.state.lock().unwrap() = RpcState::Connected;
        Ok(self.inner.multiplexer.try_broadcast(RpcState::Connected)?)
    }

    /// Try signal close to all listeners (sync)
    pub fn try_signal_close(&self) -> RpcResult<()> {
        *self.inner.state.lock().unwrap() = RpcState::Disconnected;
        Ok(self.inner.multiplexer.try_broadcast(RpcState::Disconnected)?)
    }

    /// Set the connection descriptor (URL, peer address, etc.)
    pub fn set_descriptor(&self, descriptor: Option<String>) {
        *self.inner.descriptor.lock().unwrap() = descriptor;
    }

    /// Get the connection descriptor (URL, peer address, etc.)
    pub fn descriptor(&self) -> Option<String> {
        self.inner.descriptor.lock().unwrap().clone()
    }
}