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                        let messages = match result {
215                            Ok(result) => result.data,
216                            Err(e) => {
217                                tracing::error!("Failed to fetch messages of topic: {}, {}", topic, e);
218                                continue;
219                            }
220                        };
221                        for msg in messages.iter().cloned() {
222                            yield msg
223                        }
224                        skip += messages.len();
225                    }
226                }
227            }
228        }
229    }
230
231    /// Query for swarm inspect info.
232    pub async fn inspect(&self) -> Output<SwarmInfo> {
233        let swarm_info = self
234            .client
235            .node_info(&NodeInfoRequest {})
236            .await
237            .map_err(|e| anyhow::anyhow!("{}", e))?
238            .swarm
239            .ok_or_else(|| anyhow::anyhow!("node_info response did not include swarm info"))?;
240
241        let display =
242            serde_json::to_string_pretty(&swarm_info).map_err(|e| anyhow::anyhow!("{}", e))?;
243
244        ClientOutput::ok(display, swarm_info)
245    }
246}
247
248impl<T> ClientOutput<T> {
249    /// Put display ahead to avoid moved value error.
250    pub fn ok(display: String, result: T) -> anyhow::Result<Self> {
251        Ok(Self { result, display })
252    }
253
254    /// Prints the display value of this ClientOutput instance to the console.
255    pub fn display(&self) {
256        println!("{}", self.display);
257    }
258}
rings_node/native/cli.rs

rings_node/native/
cli.rs
