rings_node/provider/
mod.rs

1#![warn(missing_docs)]
2//! General Provider, this module provide Provider implementation for FFI and WASM
3
4use std::future::Future;
5use std::pin::Pin;
6use std::sync::Arc;
7
8use rings_core::dht::EntryStorage;
9use rings_core::session::SessionSkBuilder;
10use rings_core::storage::MemStorage;
11use rings_core::swarm::callback::SharedSwarmCallback;
12use rings_rpc::protos::rings_node_handler::InternalRpcHandler;
13
14use crate::error::Error;
15use crate::error::Result;
16use crate::extension::Backend;
17use crate::measure::MeasureStorage;
18use crate::measure::PeriodicMeasure;
19use crate::prelude::wasm_export;
20use crate::processor::Processor;
21use crate::processor::ProcessorBuilder;
22use crate::processor::ProcessorConfig;
23
24#[cfg(feature = "browser")]
25pub mod browser;
26#[cfg(feature = "ffi")]
27pub mod ffi;
28
29/// General Provider, which holding reference of Processor
30/// Provider should be obey memory layout of CLang
31/// Provider should be export for wasm-bindgen
32#[derive(Clone)]
33#[allow(dead_code)]
34#[repr(C)]
35#[wasm_export]
36pub struct Provider {
37    processor: Arc<Processor>,
38    handler: InternalRpcHandler,
39    extensions: crate::extension::ext::Extensions,
40}
41
42/// Async signer, without Send required
43#[cfg(feature = "browser")]
44pub type AsyncSigner = Box<dyn Fn(String) -> Pin<Box<dyn Future<Output = Vec<u8>>>>>;
45
46/// Async signer, use for non-wasm envirement, Send is necessary
47#[cfg(not(feature = "browser"))]
48pub type AsyncSigner = Box<dyn Fn(String) -> Pin<Box<dyn Future<Output = Vec<u8>> + Send>>>;
49
50/// Signer can be async and sync
51#[allow(clippy::type_complexity)]
52pub enum Signer {
53    /// Sync signer
54    Sync(Box<dyn Fn(String) -> Vec<u8>>),
55    /// Async signer
56    Async(AsyncSigner),
57}
58
59#[allow(dead_code)]
60impl Provider {
61    /// Create provider from processor directly
62    pub fn from_processor(processor: Arc<Processor>) -> Self {
63        let extensions = crate::extension::ext::Extensions::new(processor.clone());
64        Self {
65            processor,
66            handler: InternalRpcHandler,
67            extensions,
68        }
69    }
70
71    /// The shared protocol registry. The inbound callback clones this so
72    /// registration (via the provider) and dispatch see the same table.
73    pub fn extensions(&self) -> crate::extension::ext::Extensions {
74        self.extensions.clone()
75    }
76
77    /// The capability handle — overlay `send` / `did` / self-addressed `inject`. (Authenticated
78    /// `dispatch` is router-only; `pub(crate)` so it never reaches public callers.)
79    pub(crate) fn core(&self) -> crate::extension::ext::Core {
80        self.extensions.core()
81    }
82
83    /// Register a pure [`Protocol`](crate::extension::ext::Protocol) together with its
84    /// [`Interpret`](crate::extension::ext::Interpret) shell under the protocol's namespace.
85    /// Errors if the namespace is already taken.
86    pub fn register_protocol<P, I>(&self, protocol: P, interpret: I) -> Result<()>
87    where
88        P: crate::extension::ext::Protocol + crate::extension::ext::MaybeSend + 'static,
89        P::State: crate::extension::ext::MaybeSend + 'static,
90        P::Effect: crate::extension::ext::MaybeSend,
91        I: crate::extension::ext::Interpret<Effect = P::Effect>
92            + crate::extension::ext::MaybeSend
93            + 'static,
94    {
95        self.extensions.register(protocol, interpret)
96    }
97
98    /// Send a namespaced payload to a peer. This is the uniform upper-layer send — a core
99    /// capability, identical on native and browser.
100    pub async fn send(
101        &self,
102        to: rings_core::dht::Did,
103        namespace: &str,
104        payload: bytes::Bytes,
105    ) -> Result<()> {
106        self.core().send(to, namespace, payload).await
107    }
108    /// Create a provider instance with storage name
109    pub(crate) async fn new_provider_with_storage_internal(
110        config: ProcessorConfig,
111        entry_storage: Option<EntryStorage>,
112        measure_storage: Option<MeasureStorage>,
113    ) -> Result<Provider> {
114        let entry_storage = entry_storage.unwrap_or_else(|| Box::new(MemStorage::new()));
115        let measure_storage = measure_storage.unwrap_or_else(|| Box::new(MemStorage::new()));
116
117        let measure = PeriodicMeasure::new(measure_storage);
118
119        let processor_builder = ProcessorBuilder::from_config(&config)?
120            .storage(entry_storage)
121            .measure(measure);
122
123        let processor = Arc::new(processor_builder.build()?);
124
125        let extensions = crate::extension::ext::Extensions::new(processor.clone());
126
127        Ok(Provider {
128            processor,
129            handler: InternalRpcHandler,
130            extensions,
131        })
132    }
133
134    /// Create a new provider instanice with everything in detail
135    /// Ice_servers should obey forrmat: `"[turn|strun]://<Address>:<Port>;..."`
136    /// Account is hex string
137    /// Account should format as same as account_type declared
138    /// Account_type is lowercase string, possible input are: `eip191`, `ed25519`, `bip137`, for more information,
139    /// please check [rings_core::ecc]
140    /// Signer should accept a String and returns bytes.
141    /// Signer should function as same as account_type declared, Eg: eip191 or secp256k1 or ed25519.
142    #[allow(clippy::too_many_arguments)]
143    pub(crate) async fn new_provider_internal(
144        network_id: u32,
145        ice_servers: String,
146        stabilize_interval: u64,
147        account: String,
148        account_type: String,
149        signer: Signer,
150        entry_storage: Option<EntryStorage>,
151        measure_storage: Option<MeasureStorage>,
152    ) -> Result<Provider> {
153        let mut sk_builder = SessionSkBuilder::new(account, account_type);
154        let proof = sk_builder.unsigned_proof();
155        let sig = match signer {
156            Signer::Sync(s) => s(proof),
157            Signer::Async(s) => s(proof).await,
158        };
159        sk_builder = sk_builder.set_session_sig(sig.to_vec());
160        let session_sk = sk_builder.build().map_err(Error::InternalError)?;
161        let config = ProcessorConfig::new(network_id, ice_servers, session_sk, stabilize_interval);
162        Self::new_provider_with_storage_internal(config, entry_storage, measure_storage).await
163    }
164
165    /// Install the extension [`Backend`] as the swarm's inbound callback, so inbound
166    /// custom messages are decoded as [`Envelope`](crate::extension::ext::Envelope)s and
167    /// routed to their namespace's protocol. Call once after registering protocols.
168    pub fn set_backend(&self) -> Result<()> {
169        let backend = Backend::new(Arc::new(self.clone()));
170        self.processor
171            .swarm
172            .set_callback(Arc::new(backend))
173            .map_err(Error::InternalError)
174    }
175
176    /// Set callback for swarm.
177    #[deprecated(
178        note = "set_swarm_callback will be removed in next version, plz use set_backend instead"
179    )]
180    pub fn set_swarm_callback(&self, callback: SharedSwarmCallback) -> Result<()> {
181        self.processor
182            .swarm
183            .set_callback(callback)
184            .map_err(Error::InternalError)
185    }
186
187    pub(crate) fn set_swarm_callback_internal(&self, callback: SharedSwarmCallback) -> Result<()> {
188        self.processor
189            .swarm
190            .set_callback(callback)
191            .map_err(Error::InternalError)
192    }
193
194    /// Request local rpc interface
195    /// the internal rpc interface is provide by rings_rpc
196    pub async fn request_internal(
197        &self,
198        method: String,
199        params: serde_json::Value,
200    ) -> Result<serde_json::Value> {
201        tracing::debug!("request {}", method);
202        self.handler
203            .handle_request(self.processor.clone(), method, params)
204            .await
205            .map_err(Error::InternalRpcError)
206    }
207}
208
209#[cfg(feature = "node")]
210impl Provider {
211    /// A request function implementation for native provider
212    pub async fn request<T>(
213        &self,
214        method: rings_rpc::method::Method,
215        params: T,
216    ) -> Result<serde_json::Value>
217    where
218        T: serde::Serialize,
219    {
220        let params = serde_json::to_value(params)?;
221        self.request_internal(method.to_string(), params).await
222    }
223
224    /// Listen messages
225    pub async fn listen(&self) {
226        self.processor.listen().await;
227    }
228}
rings_node/provider/mod.rs

rings_node/provider/
mod.rs
