rings_node/native/
cli.rs

1#![warn(missing_docs)]
2//! # ring-node-client
3//!
4//! ring-node-client is a command-line tool for interacting with the Ring Node backend API. It allows users to establish WebRTC connections with remote peers, send and receive messages, and publish and subscribe to topics.
5//!
6//! ## Usage
7//!
8//! To use ring-node-client, simply create a new instance of the Client struct, passing in the endpoint URL and signature as arguments. Then, use the various methods on the Client instance to perform the desired actions.
9//!
10//! # Features
11//!
12//! - Establish WebRTC connections with remote peers using HTTP as a signaling channel.
13//! - Send and receive messages using WebRTC.
14//! - Publish and subscribe to topics.
15//! - Register and lookup DIDs of services.
16//! - Send HTTP requests to remote peers.
17//! - Load a seed file to establish a connection with a remote peer.
18
19use std::time::Duration;
20
21use async_stream::stream;
22use futures::pin_mut;
23use futures::select;
24use futures::FutureExt;
25use futures::Stream;
26use futures_timer::Delay;
27use rings_rpc::jsonrpc::Client as RpcClient;
28use rings_rpc::protos::rings_node::*;
29
30use crate::seed::Seed;
31use crate::util::loader::ResourceLoader;
32
33/// Alias about `Result<ClientOutput<T>, E>`.
34type Output<T> = anyhow::Result<ClientOutput<T>>;
35
36/// Wrap json_client send request between nodes or browsers.
37pub struct Client {
38    client: RpcClient,
39}
40
41/// Wrap client output contain raw result and humanreadable display.
42pub struct ClientOutput<T> {
43    /// Output data.
44    pub result: T,
45    display: String,
46}
47
48impl Client {
49    /// Creates a new Client instance with the specified endpoint URL and signature.
50    pub fn new(endpoint_url: &str) -> anyhow::Result<Self> {
51        let rpc_client = RpcClient::new(endpoint_url);
52        Ok(Self { client: rpc_client })
53    }
54
55    /// Establishes a WebRTC connection with a remote peer using HTTP as the signaling channel.
56    ///
57    /// This function allows two peers to establish a WebRTC connection using HTTP,
58    /// which can be useful in scenarios where a direct peer-to-peer connection is not possible due to firewall restrictions or other network issues.
59    /// The function sends ICE candidates and Session Description Protocol (SDP) messages over HTTP as a form of signaling to establish the connection.
60    ///
61    /// Takes a URL for an HTTP server that will be used as the signaling channel to exchange ICE candidates and SDP with the remote peer.
62    /// Returns a Did that can be used to refer to this connection in subsequent WebRTC operations.
63    pub async fn connect_peer_via_http(&mut self, url: &str) -> Output<String> {
64        let peer_did = self
65            .client
66            .connect_peer_via_http(&ConnectPeerViaHttpRequest {
67                url: url.to_string(),
68            })
69            .await
70            .map_err(|e| anyhow::anyhow!("{}", e))?
71            .did;
72
73        ClientOutput::ok(format!("Remote did: {peer_did}"), peer_did)
74    }
75
76    /// Attempts to connect to a peer using a seed file located at the specified source path.
77    pub async fn connect_with_seed(&mut self, source: &str) -> Output<()> {
78        let seed = Seed::load(source).await?;
79        let req = seed.into_connect_with_seed_request();
80
81        self.client
82            .connect_with_seed(&req)
83            .await
84            .map_err(|e| anyhow::anyhow!("{}", e))?;
85
86        ClientOutput::ok("Successful!".to_string(), ())
87    }
88
89    /// Attempts to connect to a peer using a DID stored in a Distributed Hash Table (DHT).
90    pub async fn connect_with_did(&mut self, did: &str) -> Output<()> {
91        self.client
92            .connect_with_did(&ConnectWithDidRequest {
93                did: did.to_string(),
94            })
95            .await
96            .map_err(|e| anyhow::anyhow!("{}", e))?;
97        ClientOutput::ok("Successful!".to_owned(), ())
98    }
99
100    /// Lists all connected peers and their status.
101    ///
102    /// Returns an Output containing a formatted string representation of the list of peers if successful, or an anyhow::Error if an error occurred.
103    pub async fn list_peers(&mut self) -> Output<()> {
104        let peers = self
105            .client
106            .list_peers(&ListPeersRequest {})
107            .await
108            .map_err(|e| anyhow::anyhow!("{}", e))?
109            .peers;
110
111        let mut display = String::new();
112        display.push_str("Did, TransportId, Status\n");
113        display.push_str(
114            peers
115                .iter()
116                .map(|peer| format!("{}, {}, {}", peer.did, peer.did, peer.state))
117                .collect::<Vec<_>>()
118                .join("\n")
119                .as_str(),
120        );
121
122        ClientOutput::ok(display, ())
123    }
124
125    /// Disconnects from the peer with the specified DID.
126    pub async fn disconnect(&mut self, did: &str) -> Output<()> {
127        self.client
128            .disconnect(&DisconnectRequest {
129                did: did.to_string(),
130            })
131            .await
132            .map_err(|e| anyhow::anyhow!("{}", e))?;
133
134        ClientOutput::ok("Done.".into(), ())
135    }
136
137    /// Sends a namespaced message to the specified peer, routed to the peer's protocol
138    /// registered under `namespace` (the extension `Envelope` model).
139    pub async fn send_message(&self, did: &str, namespace: &str, data: &str) -> Output<()> {
140        self.client
141            .send_backend_message(&SendBackendMessageRequest {
142                destination_did: did.to_string(),
143                namespace: namespace.to_string(),
144                // The wire field is base64 (binary-safe); encode the raw input bytes.
145                data: base64::encode(data.as_bytes()),
146            })
147            .await
148            .map_err(|e| anyhow::anyhow!("{}", e))?;
149        ClientOutput::ok("Done.".into(), ())
150    }
151
152    /// Registers a new service with the given name.
153    pub async fn register_service(&self, name: &str) -> Output<()> {
154        self.client
155            .register_service(&RegisterServiceRequest {
156                name: name.to_string(),
157            })
158            .await
159            .map_err(|e| anyhow::anyhow!("{}", e))?;
160        ClientOutput::ok("Done.".into(), ())
161    }
162
163    /// Looks up the DIDs of services registered with the given name.
164    pub async fn lookup_service(&self, name: &str) -> Output<()> {
165        let dids = self
166            .client
167            .lookup_service(&LookupServiceRequest {
168                name: name.to_string(),
169            })
170            .await
171            .map_err(|e| anyhow::anyhow!("{}", e))?
172            .dids;
173
174        ClientOutput::ok(dids.join("\n"), ())
175    }
176
177    /// Publishes a message to the specified topic.
178    pub async fn publish_message_to_topic(&self, topic: &str, data: &str) -> Output<()> {
179        self.client
180            .publish_message_to_topic(&PublishMessageToTopicRequest {
181                topic: topic.to_string(),
182                data: data.to_string(),
183            })
184            .await
185            .map_err(|e| anyhow::anyhow!("{}", e))?;
186        ClientOutput::ok("Done.".into(), ())
187    }
188
189    /// Subscribes to the specified topic and returns a stream of messages published to the topic.
190    pub async fn subscribe_topic<'a, 'b>(
191        &'a self,
192        topic: String,
193    ) -> impl Stream<Item = String> + 'b
194    where
195        'a: 'b,
196    {
197        let mut skip = 0usize;
198
199        stream! {
200            loop {
201                let timeout = Delay::new(Duration::from_secs(5)).fuse();
202                pin_mut!(timeout);
203
204                select! {
205                    _ = timeout => {
206                        let result = self
207                            .client
208                            .fetch_topic_messages(&FetchTopicMessagesRequest {
209                                topic: topic.clone(),
210                                skip: skip as i64,
211                            })
212                            .await;
213
214                        if let Err(e) = result {
215                            tracing::error!("Failed to fetch messages of topic: {}, {}", topic, e);
216                            continue;
217                        }
218                        let messages = result.unwrap().data;
219                        for msg in messages.iter().cloned() {
220                            yield msg
221                        }
222                        skip += messages.len();
223                    }
224                }
225            }
226        }
227    }
228
229    /// Query for swarm inspect info.
230    pub async fn inspect(&self) -> Output<SwarmInfo> {
231        let swarm_info = self
232            .client
233            .node_info(&NodeInfoRequest {})
234            .await
235            .map_err(|e| anyhow::anyhow!("{}", e))?
236            .swarm
237            .unwrap();
238
239        let display =
240            serde_json::to_string_pretty(&swarm_info).map_err(|e| anyhow::anyhow!("{}", e))?;
241
242        ClientOutput::ok(display, swarm_info)
243    }
244}
245
246impl<T> ClientOutput<T> {
247    /// Put display ahead to avoid moved value error.
248    pub fn ok(display: String, result: T) -> anyhow::Result<Self> {
249        Ok(Self { result, display })
250    }
251
252    /// Prints the display value of this ClientOutput instance to the console.
253    pub fn display(&self) {
254        println!("{}", self.display);
255    }
256}
rings_node/native/cli.rs

rings_node/native/
cli.rs
