Concurrency Updates and version bump to 1.0.8

GalactechsLLC · Sep 23, 2023 · a0cc008 · a0cc008
1 parent 88837bc
commit a0cc008
Show file tree

Hide file tree

Showing 7 changed files with 73 additions and 68 deletions.
diff --git a/cli/Cargo.toml b/cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "dg_xch_cli"
-version = "1.0.7"
+version = "1.0.8"
 edition = "2021"
 authors = ["James Hoerr"]
 description = "CLI Utilities for the Chia Blockchain"
@@ -12,7 +12,7 @@ repository = "https://github.com/GalactechsLLC/dg_xch_utils/cli"
 bip39 = {version= "2.0.0", features=["rand"] }
 blst = "0.3.7"
 clap = { version = "4.1.1", features = ["derive"] }
-dg_xch_clients = {path = "../clients", version="1.0.7"}
+dg_xch_clients = {path = "../clients", version="1.0.8"}
 dg_xch_core = {path = "../core", version = "1.0.7", features = ["paperclip"] }
 dg_xch_keys = {path = "../keys", version="1.0.7"}
 dg_xch_pos = {path = "../proof_of_space", version="1.0.7"}

diff --git a/clients/Cargo.toml b/clients/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "dg_xch_clients"
-version = "1.0.7"
+version = "1.0.8"
 edition = "2021"
 authors = ["James Hoerr"]
 description = "RPC and Websocket Clients the Chia Blockchain"
@@ -10,14 +10,15 @@ repository = "https://github.com/GalactechsLLC/dg_xch_utils/clients"
 
 [dependencies]
 async-trait = "0.1.58"
+dashmap = "5.5.3"
 dg_xch_core = {path = "../core", version = "1.0.7", features = ["paperclip"] }
 dg_xch_macros = {path = "../macros", version="1.0.7"}
 dg_xch_pos = {path = "../proof_of_space", version="1.0.7"}
 dg_xch_serialize = {path = "../serialize", version="1.0.7"}
 futures-util = "0.3.25"
 hyper = {version="0.14.23", features=["full"]}
 log = "0.4.17"
-reqwest = {version="0.11.11", features=["json", "rustls-tls-webpki-roots"]}
+reqwest = {version="0.11.20", features=["json", "rustls-tls-webpki-roots"]}
 rustls = {version = "0.21.5", features = ["dangerous_configuration"] }
 rustls-pemfile = "1.0.1"
 serde = { version = "1.0.149", features = ["derive"] }

diff --git a/clients/src/websocket/farmer.rs b/clients/src/websocket/farmer.rs
@@ -6,11 +6,10 @@ use std::collections::HashMap;
 use std::io::{Error, ErrorKind};
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
-use tokio::sync::Mutex;
 use tokio::task::JoinHandle;
 
 pub struct FarmerClient {
-    pub client: Arc<Mutex<Client>>,
+    pub client: Client,
     handle: JoinHandle<()>,
 }
 impl FarmerClient {
@@ -23,8 +22,7 @@ impl FarmerClient {
     ) -> Result<Self, Error> {
         let (client, mut stream) = get_client_generated_tls(host, port, additional_headers).await?;
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
-        perform_handshake(client.clone(), network_id, port, NodeType::Farmer).await?;
+        perform_handshake(&client, network_id, port, NodeType::Farmer).await?;
         Ok(FarmerClient { client, handle })
     }
     pub async fn new_ssl(
@@ -37,8 +35,7 @@ impl FarmerClient {
     ) -> Result<Self, Error> {
         let (client, mut stream) = get_client_tls(host, port, ssl_info, additional_headers).await?;
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
-        perform_handshake(client.clone(), network_id, port, NodeType::Farmer).await?;
+        perform_handshake(&client, network_id, port, NodeType::Farmer).await?;
         Ok(FarmerClient { client, handle })
     }
     pub async fn new(
@@ -50,16 +47,15 @@ impl FarmerClient {
     ) -> Result<Self, Error> {
         let (client, mut stream) = get_client(host, port, additional_headers).await?;
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
-        perform_handshake(client.clone(), network_id, port, NodeType::Farmer).await?;
+        perform_handshake(&client, network_id, port, NodeType::Farmer).await?;
         Ok(FarmerClient { client, handle })
     }
 
-    pub async fn join(self) -> Result<(), Error> {
+    pub async fn join(mut self) -> Result<(), Error> {
         self.handle
             .await
             .map_err(|e| Error::new(ErrorKind::Other, format!("Failed to join farmer: {:?}", e)))?;
-        self.client.lock().await.shutdown().await
+        self.client.shutdown().await
     }
 
     pub fn is_closed(&self) -> bool {

diff --git a/clients/src/websocket/full_node.rs b/clients/src/websocket/full_node.rs
@@ -5,11 +5,10 @@ use std::collections::HashMap;
 use std::io::{Error, ErrorKind};
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
-use tokio::sync::Mutex;
 use tokio::task::JoinHandle;
 
 pub struct FullnodeClient {
-    pub client: Arc<Mutex<Client>>,
+    pub client: Client,
     handle: JoinHandle<()>,
 }
 impl FullnodeClient {
@@ -22,8 +21,7 @@ impl FullnodeClient {
     ) -> Result<Self, Error> {
         let (client, mut stream) = get_client(host, port, additional_headers).await?;
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
-        perform_handshake(client.clone(), network_id, port, NodeType::FullNode).await?;
+        perform_handshake(&client, network_id, port, NodeType::FullNode).await?;
         Ok(FullnodeClient { client, handle })
     }
     pub async fn new_ssl(
@@ -36,18 +34,17 @@ impl FullnodeClient {
     ) -> Result<Self, Error> {
         let (client, mut stream) = get_client_tls(host, port, ssl_info, additional_headers).await?;
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
-        perform_handshake(client.clone(), network_id, port, NodeType::FullNode).await?;
+        perform_handshake(&client, network_id, port, NodeType::FullNode).await?;
         Ok(FullnodeClient { client, handle })
     }
 
-    pub async fn join(self) -> Result<(), Error> {
+    pub async fn join(mut self) -> Result<(), Error> {
         self.handle.await.map_err(|e| {
             Error::new(
                 ErrorKind::Other,
                 format!("Failed to join fullnode: {:?}", e),
             )
         })?;
-        self.client.lock().await.shutdown().await
+        self.client.shutdown().await
     }
 }
diff --git a/clients/src/websocket/harvester.rs b/clients/src/websocket/harvester.rs
@@ -6,11 +6,10 @@ use std::collections::HashMap;
 use std::io::{Error, ErrorKind};
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
-use tokio::sync::Mutex;
 use tokio::task::JoinHandle;
 
 pub struct HarvesterClient {
-    pub client: Arc<Mutex<Client>>,
+    pub client: Client,
     handle: JoinHandle<()>,
 }
 impl HarvesterClient {
@@ -23,8 +22,7 @@ impl HarvesterClient {
     ) -> Result<Self, Error> {
         let (client, mut stream) = get_client(host, port, additional_headers).await?;
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
-        perform_handshake(client.clone(), network_id, port, NodeType::Harvester).await?;
+        perform_handshake(&client, network_id, port, NodeType::Harvester).await?;
         Ok(HarvesterClient { client, handle })
     }
     pub async fn new_ssl(
@@ -39,20 +37,19 @@ impl HarvesterClient {
         let (client, mut stream) = get_client_tls(host, port, ssl_info, additional_headers).await?;
         debug!("Spawning Stream Handler for Harvester SSL Connection");
         let handle = tokio::spawn(async move { stream.run(run).await });
-        let client = Arc::new(Mutex::new(client));
         debug!("Performing Handshake");
-        perform_handshake(client.clone(), network_id, port, NodeType::Harvester).await?;
+        perform_handshake(&client, network_id, port, NodeType::Harvester).await?;
         debug!("Harvester Handshake Complete");
         Ok(HarvesterClient { client, handle })
     }
 
-    pub async fn join(self) -> Result<(), Error> {
+    pub async fn join(mut self) -> Result<(), Error> {
         self.handle.await.map_err(|e| {
             Error::new(
                 ErrorKind::Other,
                 format!("Failed to join harvester: {:?}", e),
             )
         })?;
-        self.client.lock().await.shutdown().await
+        self.client.shutdown().await
     }
 }