redis async

Author: stneng

Cargo.toml

@@ -23,7 +23,7 @@ lapin = "2.1"
 prost = "0.10"
 rand = { version = "0.8", features = ["std_rng"] }
 rcgen = { version = "0.10", optional = true }
-redis = { version = "0.22", optional = true }
+redis = { version = "0.22", features = ["tokio-comp"], optional = true }
 reqwest = { version = "0.11", default-features = false, features = ["rustls-tls-native-roots"], optional = true }
 secp256k1 = { version = "0.25", features = ["rand-std"] }
 serde = { version = "1.0", features = ["derive"] }

src/extensions/storage_macro/redis.rs

@@ -1,23 +1,16 @@
 use async_recursion::async_recursion;
-use redis::Commands;
+use redis::{aio::Connection, AsyncCommands};
 
 type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 
 impl crate::application::CoLink {
-    fn _get_con_from_address(&self, redis_address: &str) -> Result<redis::Connection, Error> {
-        let client = redis::Client::open(redis_address)?;
-        let con = client.get_connection()?;
-        Ok(con)
-    }
-
-    async fn _get_con_from_stored_credentials(
-        &self,
-        key_path: &str,
-    ) -> Result<redis::Connection, Error> {
+    async fn _get_con_from_stored_credentials(&self, key_path: &str) -> Result<Connection, Error> {
         let redis_url_key = format!("{}:redis_url", key_path);
         let redis_url = self.read_entry(redis_url_key.as_str()).await?;
         let redis_url_string = String::from_utf8(redis_url)?;
-        self._get_con_from_address(redis_url_string.as_str())
+        let client = redis::Client::open(redis_url_string)?;
+        let con = client.get_async_connection().await?;
+        Ok(con)
     }
 
     #[async_recursion]
@@ -28,7 +21,7 @@ impl crate::application::CoLink {
         payload: &[u8],
     ) -> Result<String, Error> {
         let mut con = self._get_con_from_stored_credentials(address).await?;
-        let response: i32 = con.set_nx(key_name, payload)?;
+        let response: i32 = con.set_nx(key_name, payload).await?;
         if response == 0 {
             Err("key already exists.")?
         }
@@ -42,7 +35,7 @@ impl crate::application::CoLink {
         key_name: &str,
     ) -> Result<Vec<u8>, Error> {
         let mut con = self._get_con_from_stored_credentials(address).await?;
-        let response: Option<Vec<u8>> = con.get(key_name)?;
+        let response: Option<Vec<u8>> = con.get(key_name).await?;
         match response {
             Some(response) => Ok(response),
             None => Err("key does not exist.")?,
@@ -57,7 +50,7 @@ impl crate::application::CoLink {
         payload: &[u8],
     ) -> Result<String, Error> {
         let mut con = self._get_con_from_stored_credentials(address).await?;
-        let response = con.set(key_name, payload)?;
+        let response = con.set(key_name, payload).await?;
         Ok(response)
     }
 
@@ -70,7 +63,8 @@ impl crate::application::CoLink {
     ) -> Result<String, Error> {
         let mut con = self._get_con_from_stored_credentials(address).await?;
         let response = con
-            .append::<&str, &[u8], i32>(key_name, payload)?
+            .append::<&str, &[u8], i32>(key_name, payload)
+            .await?
             .to_string();
         Ok(response)
     }
@@ -82,7 +76,7 @@ impl crate::application::CoLink {
         key_name: &str,
     ) -> Result<String, Error> {
         let mut con = self._get_con_from_stored_credentials(address).await?;
-        let response: i32 = con.del(key_name)?;
+        let response: i32 = con.del(key_name).await?;
         if response == 0 {
             Err("key does not exist.")?
         }