Improve the API of the WebSocket Client (#2)

Still in progress, but introduces the basics of supporting typed messages.

* Wip

* Wip

* Minor

* Experiments with parsing

* Wip

* Wip

* Keep the requests ids to properly parse responses

* More cleanup

* More cleanup

Author: gmosx

xrpl_api/src/api/subscribe.rs

@@ -46,5 +46,14 @@ impl SubscribeRequest {
     }
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct LedgerClosedEvent {
+    #[serde(rename = "type")]
+    pub event_type: String,
+    pub fee_base: u32,
+    pub fee_ref: u32,
+    pub txn_count: u32,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct SubscribeResponse {}

xrpl_sdk_ws/Cargo.toml

@@ -12,6 +12,7 @@ edition = "2021"
 thiserror = "1"
 serde = { version = "1", features = ["derive"]}
 serde_json = "1"
+futures = "0.3"
 futures-util = "0.3"
 tokio = { version = "1", features = ["full"] }
 tokio-stream = "0.1"

xrpl_sdk_ws/src/client.rs

@@ -1,12 +1,15 @@
 use crate::util::Result;
+use futures::{future, stream::SplitSink, StreamExt};
 use futures_util::SinkExt;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
+use std::{cell::RefCell, collections::HashMap, pin::Pin, rc::Rc};
 use tokio::net::TcpStream;
+use tokio_stream::Stream;
 use tokio_tungstenite::{
     connect_async, tungstenite::protocol::Message, MaybeTlsStream, WebSocketStream,
 };
 use uuid::Uuid;
-use xrpl_api::Request;
+use xrpl_api::{AccountInfoResponse, LedgerClosedEvent, Request};
 
 // https://xrpl.org/public-servers.html
 
@@ -21,15 +24,71 @@ pub const DEFAULT_WS_URL: &str = XRPL_CLUSTER_MAINNET_WS_URL;
 
 // #TODO extract Connection
 
+#[derive(Serialize, Deserialize, Debug)]
+pub enum TypedMessage {
+    AccountInfo(AccountInfoResponse),
+    LedgerClosed(LedgerClosedEvent),
+    Other(String),
+}
+
 /// A WebSocket client for the XRP Ledger.
 pub struct Client {
-    pub stream: WebSocketStream<MaybeTlsStream<TcpStream>>,
+    sender: SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
+    requests: Rc<RefCell<HashMap<String, String>>>,
+    pub messages: Pin<Box<dyn Stream<Item = Result<TypedMessage>>>>,
 }
 
 impl Client {
     pub async fn connect(url: &str) -> Result<Self> {
         let (stream, _response) = connect_async(url).await?;
-        Ok(Self { stream })
+        let (sender, receiver) = stream.split();
+        let requests: Rc<RefCell<HashMap<String, String>>> = Rc::new(RefCell::new(HashMap::new()));
+
+        let cloned_requests = requests.clone();
+        let receiver = receiver
+            .map(move |msg| {
+                if let Message::Text(string) = msg.unwrap() {
+                    let mut value: serde_json::Value = serde_json::from_str(&string).unwrap();
+
+                    if let Some(id) = value["id"].as_str() {
+                        // If the message contains an id field it's a response to
+                        // an RPC request.
+                        if let Some(method) = requests.borrow_mut().get(id) {
+                            let result = value["result"].take();
+                            match method.as_str() {
+                                "account_info" => Ok(Some(TypedMessage::AccountInfo(
+                                    serde_json::from_value(result)?,
+                                ))),
+                                _ => Ok(Some(TypedMessage::Other(string))),
+                            }
+                        } else {
+                            Ok(Some(TypedMessage::Other(string)))
+                        }
+                    } else {
+                        // If the message has no id field, it's a subscription event.
+
+                        if let Some(event_type) = value["type"].as_str() {
+                            match event_type {
+                                "ledgerClosed" => Ok(Some(TypedMessage::LedgerClosed(
+                                    serde_json::from_value(value)?,
+                                ))),
+                                _ => Ok(Some(TypedMessage::Other(string))),
+                            }
+                        } else {
+                            Ok(Some(TypedMessage::Other(string)))
+                        }
+                    }
+                } else {
+                    Ok(None)
+                }
+            })
+            .filter_map(|res| future::ready(res.transpose()));
+
+        Ok(Self {
+            sender,
+            messages: Box::pin(receiver),
+            requests: cloned_requests,
+        })
     }
 
     pub async fn call<Req>(&mut self, req: Req) -> Result<()>
@@ -43,14 +102,16 @@ impl Client {
         // #TODO, this is temp code, add error-handling!
 
         if let serde_json::Value::Object(mut map) = msg {
-            map.insert("id".to_owned(), serde_json::Value::String(id));
+            map.insert("id".to_owned(), serde_json::Value::String(id.clone()));
             map.insert(
                 "command".to_owned(),
                 serde_json::Value::String(req.method()),
             );
             let msg = serde_json::to_string(&map).unwrap();
 
-            self.stream.send(Message::Text(msg.to_string())).await?;
+            self.sender.send(Message::Text(msg.to_string())).await?;
+
+            self.requests.borrow_mut().insert(id, req.method());
         }
 
         Ok(())

xrpl_sdk_ws/src/client_tests.rs

@@ -15,11 +15,7 @@ mod tests {
 
         client.call(req).await.expect("cannot send request");
 
-        let (_, rx) = client.stream.split();
-
-        tokio::pin!(rx);
-
-        while let Some(msg) = rx.next().await {
+        if let Some(msg) = client.messages.next().await {
             dbg!(&msg);
         }
     }
@@ -31,14 +27,17 @@ mod tests {
             .expect("cannot connect");
 
         let req = SubscribeRequest::streams(&["ledger"]);
-        client.call(req).await.expect("cannot subscribe");
 
-        let (_, rx) = client.stream.split();
+        client.call(req).await.expect("cannot subscribe");
 
-        tokio::pin!(rx);
+        let mut count = 0;
 
-        while let Some(msg) = rx.next().await {
+        while let Some(msg) = client.messages.next().await {
+            if count > 2 {
+                break;
+            }
             dbg!(&msg);
+            count += 1;
         }
     }
 }

xrpl_sdk_ws/src/error.rs

@@ -7,20 +7,18 @@ use thiserror::Error;
 pub enum Error {
     #[error("internal error: {0}")]
     Internal(String),
-    // #[error("failed request: {err}")]
-    // FailedRequest { err: String, status: Option<u16> },
-    // #[error("not authorized: missing api_credentials")]
-    // Unauthorized,
-    // #[error("api error: {0}")]
-    // Api(String),
+    #[error("malformed JSON payload: {0}")]
+    MalformedJSON(String),
 }
 
 impl From<tokio_tungstenite::tungstenite::Error> for Error {
     fn from(e: tokio_tungstenite::tungstenite::Error) -> Self {
         Self::Internal(e.to_string())
-        // Self::Internal {
-        //     err: e.to_string(),
-        //     status: e.status().map(|c| c.as_u16()),
-        // }
+    }
+}
+
+impl From<serde_json::Error> for Error {
+    fn from(e: serde_json::Error) -> Self {
+        Self::MalformedJSON(e.to_string())
     }
 }