fix: Future is not send

bcnmy · Jul 11, 2024 · 2b70edf · 2b70edf
1 parent cf0392c
commit 2b70edf
Show file tree

Hide file tree

Showing 10 changed files with 79 additions and 75 deletions.
diff --git a/crates/routing-engine/src/indexer.rs b/crates/routing-engine/src/indexer.rs
@@ -308,6 +308,7 @@ mod tests {
     use std::sync::Arc;
     use std::time::Duration;
 
+    use async_trait::async_trait;
     use derive_more::Display;
     use thiserror::Error;
 
@@ -324,6 +325,8 @@ mod tests {
 
     #[derive(Debug)]
     struct ModelStoreStub;
+
+    #[async_trait]
     impl KeyValueStore for ModelStoreStub {
         type Error = Err;
 
@@ -354,6 +357,8 @@ mod tests {
 
     #[derive(Debug)]
     struct ProducerStub;
+
+    #[async_trait]
     impl MessageQueue for ProducerStub {
         type Error = Err;
 
@@ -372,6 +377,8 @@ mod tests {
 
     #[derive(Debug)]
     struct TokenPriceProviderStub;
+
+    #[async_trait]
     impl TokenPriceProvider for TokenPriceProviderStub {
         type Error = Error;
 

diff --git a/crates/routing-engine/src/settlement_engine.rs b/crates/routing-engine/src/settlement_engine.rs
@@ -7,7 +7,6 @@ use alloy::transports::http::Http;
 use futures::StreamExt;
 use log::{error, info};
 use reqwest::{Client, Url};
-use ruint::aliases::U256;
 use ruint::Uint;
 use serde::Serialize;
 use thiserror::Error;
@@ -72,31 +71,28 @@ impl<Source: RouteSource, PriceProvider: TokenPriceProvider>
             .map(|route| async move {
                 info!("Generating transactions for route: {:?}", route.route);
 
-                // let token_amount = get_token_amount_from_value_in_usd(
-                //     &self.config,
-                //     &self.price_provider,
-                //     &route.route.from_token.symbol,
-                //     route.route.from_chain.id,
-                //     &route.source_amount_in_usd,
-                // )
-                // .await
-                // .map_err(|err| SettlementEngineErrors::GetTokenAmountFromValueInUsdError(err))?;
-                let token_amount: U256 = Uint::from(100);
+                let token_amount = get_token_amount_from_value_in_usd(
+                    &self.config,
+                    &self.price_provider,
+                    &route.route.from_token.symbol,
+                    route.route.from_chain.id,
+                    &route.source_amount_in_usd,
+                )
+                .await
+                .map_err(|err| SettlementEngineErrors::GetTokenAmountFromValueInUsdError(err))?;
 
                 info!("Token amount: {:?} for route {:?}", token_amount, route);
 
-                // let (ethereum_transactions, required_approval_details) = self
-                //     .source
-                //     .generate_route_transactions(
-                //         &route.route,
-                //         &token_amount,
-                //         &route.from_address,
-                //         &route.to_address,
-                //     )
-                //     .await
-                //     .map_err(|err| SettlementEngineErrors::GenerateTransactionsError(err))?;
-                let ethereum_transactions = Vec::new();
-                let required_approval_details = Vec::new();
+                let (ethereum_transactions, required_approval_details) = self
+                    .source
+                    .generate_route_transactions(
+                        &route.route,
+                        &token_amount,
+                        &route.from_address,
+                        &route.to_address,
+                    )
+                    .await
+                    .map_err(|err| SettlementEngineErrors::GenerateTransactionsError(err))?;
 
                 info!("Generated transactions: {:?} for route {:?}", ethereum_transactions, route);
 
@@ -392,6 +388,7 @@ mod tests {
     use std::time::Duration;
 
     use alloy::primitives::U256;
+    use async_trait::async_trait;
     use derive_more::Display;
     use thiserror::Error;
 
@@ -412,6 +409,7 @@ mod tests {
         map: Mutex<HashMap<String, String>>,
     }
 
+    #[async_trait]
     impl KeyValueStore for KVStore {
         type Error = Err;
 
@@ -750,5 +748,12 @@ mod tests {
         assert_is_send::<EthereumTransaction>();
         assert_is_send::<RequiredApprovalDetails>();
         assert_is_send::<SettlementEngineErrors<BungeeClient, CoingeckoClient<KVStore>>>();
+        assert_is_send::<U256>();
+        assert_is_send::<
+            Result<
+                (Vec<EthereumTransaction>, Vec<RequiredApprovalDetails>),
+                SettlementEngineErrors<BungeeClient, CoingeckoClient<KVStore>>,
+            >,
+        >()
     }
 }
diff --git a/crates/routing-engine/src/source/bungee/mod.rs b/crates/routing-engine/src/source/bungee/mod.rs
@@ -1,5 +1,6 @@
 use std::str::FromStr;
 
+use async_trait::async_trait;
 use log::{error, info};
 use reqwest;
 use reqwest::header;
@@ -106,6 +107,7 @@ pub enum GenerateRouteTransactionsError {
     InvalidU256Error(String),
 }
 
+#[async_trait]
 impl RouteSource for BungeeClient {
     type FetchRouteCostError = BungeeFetchRouteCostError;
 

diff --git a/crates/routing-engine/src/source/mod.rs b/crates/routing-engine/src/source/mod.rs
@@ -1,6 +1,7 @@
 use std::error::Error;
 use std::fmt::Debug;
 
+use async_trait::async_trait;
 use ruint::aliases::U256;
 use serde::Serialize;
 
@@ -25,30 +26,29 @@ pub struct RequiredApprovalDetails {
     pub amount: U256,
 }
 
+#[async_trait]
 pub trait RouteSource: Debug + Send + Sync {
     type FetchRouteCostError: Debug + Error + Send + Sync;
     type GenerateRouteTransactionsError: Debug + Error + Send + Sync;
     type BaseRouteType: Debug + Send + Sync;
 
-    fn fetch_least_cost_route_and_cost_in_usd(
+    async fn fetch_least_cost_route_and_cost_in_usd(
         &self,
         route: &Route,
         from_token_amount: &U256,
         sender_address: Option<&String>,
         recipient_address: Option<&String>,
         estimation_type: &CostType,
-    ) -> impl futures::Future<Output = Result<(Self::BaseRouteType, f64), Self::FetchRouteCostError>>;
+    ) -> Result<(Self::BaseRouteType, f64), Self::FetchRouteCostError>;
 
-    fn generate_route_transactions(
+    async fn generate_route_transactions(
         &self,
         route: &Route,
         amount: &U256,
         sender_address: &String,
         recipient_address: &String,
-    ) -> impl futures::Future<
-        Output = Result<
-            (Vec<EthereumTransaction>, Vec<RequiredApprovalDetails>),
-            Self::GenerateRouteTransactionsError,
-        >,
+    ) -> Result<
+        (Vec<EthereumTransaction>, Vec<RequiredApprovalDetails>),
+        Self::GenerateRouteTransactionsError,
     >;
 }
diff --git a/crates/routing-engine/src/token_price/coingecko.rs b/crates/routing-engine/src/token_price/coingecko.rs
@@ -2,6 +2,7 @@ use std::fmt::Debug;
 use std::num::ParseFloatError;
 use std::time::Duration;
 
+use async_trait::async_trait;
 use derive_more::Display;
 use log::{error, info};
 use reqwest::{header, StatusCode};
@@ -70,6 +71,7 @@ impl<KVStore: KeyValueStore> CoingeckoClient<KVStore> {
     }
 }
 
+#[async_trait]
 impl<KVStore: KeyValueStore> TokenPriceProvider for CoingeckoClient<KVStore> {
     type Error = CoingeckoClientError<KVStore>;
 
@@ -140,6 +142,7 @@ mod tests {
     use std::sync::Mutex;
     use std::time::Duration;
 
+    use async_trait::async_trait;
     use derive_more::Display;
     use thiserror::Error;
 
@@ -157,6 +160,7 @@ mod tests {
         map: Mutex<HashMap<String, String>>,
     }
 
+    #[async_trait]
     impl KeyValueStore for KVStore {
         type Error = Err;
 

diff --git a/crates/routing-engine/src/token_price/mod.rs b/crates/routing-engine/src/token_price/mod.rs
@@ -1,16 +1,16 @@
 use std::error::Error;
 use std::fmt::Debug;
 
+use async_trait::async_trait;
+
 pub use coingecko::CoingeckoClient;
 
 mod coingecko;
 pub mod utils;
 
+#[async_trait]
 pub trait TokenPriceProvider: Debug + Send + Sync {
     type Error: Error + Debug + Send + Sync;
 
-    fn get_token_price(
-        &self,
-        token_symbol: &String,
-    ) -> impl futures::Future<Output = Result<f64, Self::Error>>;
+    async fn get_token_price(&self, token_symbol: &String) -> Result<f64, Self::Error>;
 }
diff --git a/crates/routing-engine/src/token_price/utils.rs b/crates/routing-engine/src/token_price/utils.rs
@@ -71,6 +71,7 @@ pub enum Errors<T: Debug + Send + Sync> {
 mod tests {
     use std::fmt::Error;
 
+    use async_trait::async_trait;
     use ruint::Uint;
 
     use config::{Config, get_sample_config};
@@ -84,6 +85,7 @@ mod tests {
     #[derive(Debug)]
     struct TokenPriceProviderStub;
 
+    #[async_trait]
     impl TokenPriceProvider for TokenPriceProviderStub {
         type Error = Error;
 

diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs
@@ -1,10 +1,10 @@
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt::Debug;
-use std::future;
 use std::time::Duration;
 
 pub use ::redis::{ControlFlow, Msg};
+use async_trait::async_trait;
 use mongodb::bson::Document;
 
 pub use redis_client::{RedisClient, RedisClientError};
@@ -13,43 +13,28 @@ pub mod mongodb_client;
 
 mod redis_client;
 
+#[async_trait]
 pub trait KeyValueStore: Debug + Send + Sync {
     type Error: Error + Debug + Send + Sync;
 
-    fn get(&self, k: &String) -> impl future::Future<Output = Result<String, Self::Error>>;
+    async fn get(&self, k: &String) -> Result<String, Self::Error>;
 
-    fn get_multiple(
-        &self,
-        k: &Vec<String>,
-    ) -> impl future::Future<Output = Result<Vec<String>, Self::Error>>;
+    async fn get_multiple(&self, k: &Vec<String>) -> Result<Vec<String>, Self::Error>;
 
-    fn set(
-        &self,
-        k: &String,
-        v: &String,
-        expiry: Duration,
-    ) -> impl future::Future<Output = Result<(), Self::Error>>;
+    async fn set(&self, k: &String, v: &String, expiry: Duration) -> Result<(), Self::Error>;
 
-    fn set_multiple(
-        &self,
-        kv: &Vec<(String, String)>,
-    ) -> impl future::Future<Output = Result<(), Self::Error>>;
+    async fn set_multiple(&self, kv: &Vec<(String, String)>) -> Result<(), Self::Error>;
 
-    fn get_all_keys(&self) -> impl future::Future<Output = Result<Vec<String>, RedisClientError>>;
+    async fn get_all_keys(&self) -> Result<Vec<String>, RedisClientError>;
 
-    fn get_all_key_values(
-        &self,
-    ) -> impl future::Future<Output = Result<HashMap<String, String>, RedisClientError>>;
+    async fn get_all_key_values(&self) -> Result<HashMap<String, String>, RedisClientError>;
 }
 
-pub trait MessageQueue: Debug {
-    type Error: Error + Debug;
+#[async_trait]
+pub trait MessageQueue: Debug + Send + Sync {
+    type Error: Error + Debug + Send + Sync;
 
-    fn publish(
-        &self,
-        topic: &str,
-        message: &str,
-    ) -> impl future::Future<Output = Result<(), Self::Error>>;
+    async fn publish(&self, topic: &str, message: &str) -> Result<(), Self::Error>;
 
     fn subscribe<U>(
         &self,
@@ -58,21 +43,15 @@ pub trait MessageQueue: Debug {
     ) -> Result<(), Self::Error>;
 }
 
-pub trait DBProvider: Debug {
-    type Error: Error + Debug;
+#[async_trait]
+pub trait DBProvider: Debug + Send + Sync {
+    type Error: Error + Debug + Send + Sync;
 
-    fn create(&self, item: &Document) -> impl future::Future<Output = Result<(), Self::Error>>;
+    async fn create(&self, item: &Document) -> Result<(), Self::Error>;
 
-    fn read(
-        &self,
-        query: &Document,
-    ) -> impl future::Future<Output = Result<Option<Document>, Self::Error>>;
+    async fn read(&self, query: &Document) -> Result<Option<Document>, Self::Error>;
 
-    fn update(
-        &self,
-        query: &Document,
-        update: &Document,
-    ) -> impl future::Future<Output = Result<(), Self::Error>>;
+    async fn update(&self, query: &Document, update: &Document) -> Result<(), Self::Error>;
 
-    fn delete(&self, query: &Document) -> impl future::Future<Output = Result<(), Self::Error>>;
+    async fn delete(&self, query: &Document) -> Result<(), Self::Error>;
 }
diff --git a/crates/storage/src/mongodb_client.rs b/crates/storage/src/mongodb_client.rs
@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use derive_more::Display;
 use mongodb::{
     bson::{self, doc, Document},
@@ -65,6 +66,7 @@ impl MongoDBClient {
     }
 }
 
+#[async_trait]
 impl DBProvider for MongoDBClient {
     type Error = DBError;
 

diff --git a/crates/storage/src/redis_client.rs b/crates/storage/src/redis_client.rs
@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::time::Duration;
 
+use async_trait::async_trait;
 use log::info;
 use redis::{self, aio, AsyncCommands, ControlFlow, Msg, PubSubCommands};
 use redis::RedisError;
@@ -22,6 +23,7 @@ impl RedisClient {
     }
 }
 
+#[async_trait]
 impl KeyValueStore for RedisClient {
     type Error = RedisClientError;
 
@@ -65,6 +67,7 @@ impl KeyValueStore for RedisClient {
     }
 }
 
+#[async_trait]
 impl MessageQueue for RedisClient {
     type Error = RedisClientError;