async thread safe

Signed-off-by: Sam Batschelet <[email protected]>

Author: hexfusion

spaces-cli/src/bin/spaces-cli/main.rs

@@ -49,19 +49,22 @@ async fn main() -> Result<(), Box<dyn error::Error>> {
 
     let private_key = get_or_create_pk(&cli.private_key_file)?;
     let uri = cli.endpoint.parse::<Uri>()?;
-    let mut client = Client::new(uri).set_private_key(private_key);
+    let client = Client::new(uri);
+    client.set_private_key(private_key).await;
 
     if let Command::Get { space, key } = &cli.command {
-        let resp =
-            futures::executor::block_on(client.resolve(space, key)).map_err(|e| e.to_string())?;
+        let resp = client
+            .resolve(space, key)
+            .await
+            .map_err(|e| e.to_string())?;
         log::debug!("resolve response: {:?}", resp);
 
         println!("{}", serde_json::to_string(&resp)?);
         return Ok(());
     }
 
     if let Command::Ping {} = &cli.command {
-        let resp = futures::executor::block_on(client.ping()).map_err(|e| e.to_string())?;
+        let resp = client.ping().await.map_err(|e| e.to_string())?;
 
         println!("{}", serde_json::to_string(&resp)?);
         return Ok(());
@@ -74,8 +77,10 @@ async fn main() -> Result<(), Box<dyn error::Error>> {
     let typed_data = &resp.typed_data;
 
     // issue tx
-    let resp =
-        futures::executor::block_on(client.issue_tx(typed_data)).map_err(|e| e.to_string())?;
+    let resp = client
+        .issue_tx(typed_data)
+        .await
+        .map_err(|e| e.to_string())?;
     println!("{}", serde_json::to_string(&resp)?);
 
     Ok(())

spacesvm/Cargo.toml

@@ -15,7 +15,7 @@ name = "spacesvm"
 path = "src/bin/spaces/main.rs"
 
 [dependencies]
-avalanche-types = { version = "0.0.140", features = ["subnet"] }
+avalanche-types = { version = "0.0.143", features = ["subnet"] }
 byteorder = "1.4.3"
 chrono = "0.4.22"
 crossbeam-channel = "0.5.6"

spacesvm/src/api/client.rs

@@ -2,6 +2,7 @@ use std::{
     fs::File,
     io::{Error, ErrorKind, Result, Write},
     path::Path,
+    sync::Arc,
 };
 
 use crate::{
@@ -25,9 +26,14 @@ use jsonrpc_core::{Call, Id, MethodCall, Params, Value, Version};
 use serde::de;
 
 pub use http::Uri;
+use tokio::sync::RwLock;
 
-/// HTTP client for interacting with the API, assumes single threaded use.
+/// HTTP client for interacting with the API.
 pub struct Client<C> {
+    inner: Arc<RwLock<ClientInner<C>>>,
+}
+
+pub struct ClientInner<C> {
     id: u64,
     client: HyperClient<C>,
     endpoint: Uri,
@@ -38,33 +44,37 @@ impl Client<HttpConnector> {
     pub fn new(endpoint: Uri) -> Self {
         let client = HyperClient::new();
         Self {
-            id: 0,
-            client,
-            endpoint,
-            private_key: None,
+            inner: Arc::new(RwLock::new(ClientInner {
+                id: 0,
+                client,
+                endpoint,
+                private_key: None,
+            })),
         }
     }
 }
 
 impl Client<HttpConnector> {
-    fn next_id(&mut self) -> Id {
-        let id = self.id;
-        self.id = id + 1;
+    async fn next_id(&self) -> Id {
+        let mut client = self.inner.write().await;
+        let id = client.id;
+        client.id = id + 1;
         Id::Num(id)
     }
 
-    pub fn set_endpoint(&mut self, endpoint: Uri) {
-        self.endpoint = endpoint;
+    pub async fn set_endpoint(&self, endpoint: Uri) {
+        let mut inner = self.inner.write().await;
+        inner.endpoint = endpoint;
     }
 
-    pub fn set_private_key(mut self, private_key: Key) -> Self {
-        self.private_key = Some(private_key);
-        self
+    pub async fn set_private_key(&self, private_key: Key) {
+        let mut inner = self.inner.write().await;
+        inner.private_key = Some(private_key);
     }
 
     /// Returns a serialized json request as string and the request id.
-    pub fn raw_request(&mut self, method: &str, params: &Params) -> (Id, String) {
-        let id = self.next_id();
+    pub async fn raw_request(&self, method: &str, params: &Params) -> (Id, String) {
+        let id = self.next_id().await;
         let request = jsonrpc_core::Request::Single(Call::MethodCall(MethodCall {
             jsonrpc: Some(Version::V2),
             method: method.to_owned(),
@@ -77,64 +87,73 @@ impl Client<HttpConnector> {
         )
     }
 
-    /// Returns a recoverable signature from bytes.
-    pub fn sign_digest(&self, dh: &[u8]) -> Result<Sig> {
-        if let Some(pk) = &self.private_key {
+    /// Returns a recoverable signature from 32 byte SHA256 message.
+    pub async fn sign_digest(&self, dh: &[u8]) -> Result<Sig> {
+        let inner = self.inner.read().await;
+        if let Some(pk) = &inner.private_key {
             return pk.sign_digest(dh);
         }
         Err(Error::new(ErrorKind::Other, "private key not set"))
     }
 
     /// Returns a PingResponse from client request.
-    pub async fn ping(&mut self) -> Result<PingResponse> {
-        let (_id, json_request) = self.raw_request("ping", &Params::None);
+    pub async fn ping(&self) -> Result<PingResponse> {
+        let (_id, json_request) = self.raw_request("ping", &Params::None).await;
         let resp = self.post_de::<PingResponse>(&json_request).await?;
 
         Ok(resp)
     }
 
     /// Returns a DecodeTxResponse from client request.
-    pub async fn decode_tx(&mut self, tx_data: TransactionData) -> Result<DecodeTxResponse> {
+    pub async fn decode_tx(&self, tx_data: TransactionData) -> Result<DecodeTxResponse> {
         let arg_value = serde_json::to_value(&DecodeTxArgs { tx_data })?;
-        let (_id, json_request) = self.raw_request("decodeTx", &Params::Array(vec![arg_value]));
+        let (_id, json_request) = self
+            .raw_request("decodeTx", &Params::Array(vec![arg_value]))
+            .await;
         let resp = self.post_de::<DecodeTxResponse>(&json_request).await?;
 
         Ok(resp)
     }
 
     /// Returns a IssueTxResponse from client request.
-    pub async fn issue_tx(&mut self, typed_data: &TypedData) -> Result<IssueTxResponse> {
+    pub async fn issue_tx(&self, typed_data: &TypedData) -> Result<IssueTxResponse> {
         let dh = decoder::hash_structured_data(typed_data)?;
-        let sig = self.sign_digest(&dh.as_bytes())?.to_bytes().to_vec();
+        let sig = self.sign_digest(&dh.as_bytes()).await?.to_bytes().to_vec();
         log::debug!("signature: {:?}", sig);
 
         let arg_value = serde_json::to_value(&IssueTxArgs {
             typed_data: typed_data.to_owned(),
             signature: sig,
         })?;
-        let (_id, json_request) = self.raw_request("issueTx", &Params::Array(vec![arg_value]));
+        let (_id, json_request) = self
+            .raw_request("issueTx", &Params::Array(vec![arg_value]))
+            .await;
         let resp = self.post_de::<IssueTxResponse>(&json_request).await?;
 
         Ok(resp)
     }
 
     /// Returns a ResolveResponse from client request.
-    pub async fn resolve(&mut self, space: &str, key: &str) -> Result<ResolveResponse> {
+    pub async fn resolve(&self, space: &str, key: &str) -> Result<ResolveResponse> {
         let arg_value = serde_json::to_value(&ResolveArgs {
             space: space.as_bytes().to_vec(),
             key: key.as_bytes().to_vec(),
         })?;
-        let (_id, json_request) = self.raw_request("resolve", &Params::Array(vec![arg_value]));
+        let (_id, json_request) = self
+            .raw_request("resolve", &Params::Array(vec![arg_value]))
+            .await;
         let resp = self.post_de::<ResolveResponse>(&json_request).await?;
 
         Ok(resp)
     }
 
     /// Returns a deserialized response from client request.
     pub async fn post_de<T: de::DeserializeOwned>(&self, json: &str) -> Result<T> {
+        let inner = self.inner.read().await;
+
         let req = Request::builder()
             .method(Method::POST)
-            .uri(self.endpoint.to_string())
+            .uri(inner.endpoint.to_string())
             .header("content-type", "application/json-rpc")
             .body(Body::from(json.to_owned()))
             .map_err(|e| {
@@ -144,7 +163,7 @@ impl Client<HttpConnector> {
                 )
             })?;
 
-        let mut resp = self.client.request(req).await.map_err(|e| {
+        let mut resp = inner.client.request(req).await.map_err(|e| {
             std::io::Error::new(
                 std::io::ErrorKind::Other,
                 format!("client post request failed: {}", e),
@@ -219,12 +238,12 @@ pub fn get_or_create_pk(path: &str) -> Result<key::secp256k1::private_key::Key>
         .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))
 }
 
-#[test]
-fn test_raw_request() {
-    let mut cli = Client::new(Uri::from_static("http://test.url"));
-    let (id, _) = cli.raw_request("ping", &Params::None);
+#[tokio::test]
+async fn test_raw_request() {
+    let cli = Client::new(Uri::from_static("http://test.url"));
+    let (id, _) = cli.raw_request("ping", &Params::None).await;
     assert_eq!(id, jsonrpc_core::Id::Num(0));
-    let (id, req) = cli.raw_request("ping", &Params::None);
+    let (id, req) = cli.raw_request("ping", &Params::None).await;
     assert_eq!(id, jsonrpc_core::Id::Num(1));
     assert_eq!(
         req,

spacesvm/src/bin/spaces/main.rs

@@ -45,7 +45,7 @@ async fn main() -> Result<()> {
     ) = tokio::sync::broadcast::channel(1);
 
     info!("starting spacesvm-rs");
-    let vm_server = subnet::rpc::vm::server::Server::new(Box::new(vm::ChainVm::new()), stop_ch_tx);
+    let vm_server = subnet::rpc::vm::server::Server::new(vm::ChainVm::new(), stop_ch_tx);
 
     subnet::rpc::plugin::serve(vm_server, stop_ch_rx)
         .await

spacesvm/tests/vm/mod.rs

@@ -26,10 +26,8 @@ async fn test_api() {
 
     // setup stop channel for grpc services.
     let (stop_ch_tx, stop_ch_rx): (Sender<()>, Receiver<()>) = tokio::sync::broadcast::channel(1);
-    let vm_server = avalanche_types::subnet::rpc::vm::server::Server::new(
-        Box::new(vm::ChainVm::new()),
-        stop_ch_tx,
-    );
+    let vm_server =
+        avalanche_types::subnet::rpc::vm::server::Server::new(Box::new(vm::ChainVm::new()), stop_ch_tx);
 
     // start Vm service
     let vm_addr = utils::new_socket_addr();
@@ -115,7 +113,7 @@ async fn test_api() {
     let mut client = spacesvm::api::client::Client::new(http::Uri::from_static("http://test.url"));
 
     // ping
-    let (_id, json_str) = client.raw_request("ping", &Params::None);
+    let (_id, json_str) = client.raw_request("ping", &Params::None).await;
     let req = http::request::Builder::new()
         .body(json_str.as_bytes().to_vec())
         .unwrap();
@@ -141,7 +139,9 @@ async fn test_api() {
     let tx_data = claim_tx("test_claim");
     let arg_value = serde_json::to_value(&DecodeTxArgs { tx_data }).unwrap();
 
-    let (_id, json_str) = client.raw_request("decodeTx", &Params::Array(vec![arg_value]));
+    let (_id, json_str) = client
+        .raw_request("decodeTx", &Params::Array(vec![arg_value]))
+        .await;
     log::info!("decodeTx request: {}", json_str);
     let req = http::request::Builder::new()
         .body(json_str.as_bytes().to_vec())

tests/e2e/src/tests/mod.rs

@@ -194,22 +194,23 @@ async fn e2e() {
 
     let private_key = get_or_create_pk("/tmp/.spacesvm-cli-pk").expect("generate new private key");
     let chain_url = format!("{}/ext/bc/{}/public", rpc_eps[0], blockchain_id);
-    let mut scli =
-        spacesvm::api::client::Client::new(chain_url.parse::<Uri>().expect("valid endpoint"))
-            .set_private_key(private_key);
+    let scli =
+        spacesvm::api::client::Client::new(chain_url.parse::<Uri>().expect("valid endpoint"));
+    scli.set_private_key(private_key).await;
     for ep in rpc_eps.iter() {
         let chain_url = format!("{}/ext/bc/{}/public", ep, blockchain_id)
             .parse::<Uri>()
             .expect("valid endpoint");
-        scli.set_endpoint(chain_url);
+        scli.set_endpoint(chain_url).await;
         let resp = scli.ping().await.unwrap();
         log::info!("ping response from {}: {:?}", ep, resp);
         assert!(resp.success);
 
         thread::sleep(Duration::from_millis(300));
     }
 
-    scli.set_endpoint(chain_url.parse::<Uri>().expect("valid endpoint"));
+    scli.set_endpoint(chain_url.parse::<Uri>().expect("valid endpoint"))
+        .await;
 
     log::info!("decode claim tx request...");
     let resp = scli