rings_node/native/
cli.rs

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

rings_node/native/
cli.rs
