feat(sdk): reuse dapi connections

packages/rs-dapi-client/Cargo.toml

@@ -24,6 +24,7 @@ tokio = { version = "1.32.0", default-features = false }
 sha2 = { version = "0.10", optional = true }
 chrono = { version = "0.4.31", optional = true }
 hex = { version = "0.4.3", optional = true }
+lru = { version = "0.12.3" }
 
 [dev-dependencies]
 tokio = { version = "1.32.0", features = ["macros"] }

packages/rs-dapi-client/src/address_list.rs

@@ -160,17 +160,41 @@ impl AddressList {
 
     /// Randomly select a not banned address.
     pub fn get_live_address(&self) -> Option<&Address> {
-        let now = time::Instant::now();
         let mut rng = SmallRng::from_entropy();
 
+        self.unbanned().into_iter().choose(&mut rng)
+    }
+
+    /// Get all addresses that are not banned.
+    fn unbanned(&self) -> Vec<&Address> {
+        let now = time::Instant::now();
+
         self.addresses
             .iter()
             .filter(|addr| {
                 addr.banned_until
                     .map(|banned_until| banned_until < now)
                     .unwrap_or(true)
             })
-            .choose(&mut rng)
+            .collect()
+    }
+
+    /// Get number of available, not banned addresses.
+    pub fn available(&self) -> usize {
+        self.unbanned().len()
+    }
+
+    /// Get number of all addresses, both banned and not banned.
+    pub fn len(&self) -> usize {
+        self.addresses.len()
+    }
+
+    /// Check if the list is empty.
+    /// Returns true if there are no addresses in the list.
+    /// Returns false if there is at least one address in the list.
+    /// Banned addresses are also counted.
+    pub fn is_empty(&self) -> bool {
+        self.addresses.is_empty()
     }
 }
 

packages/rs-dapi-client/src/connection_pool.rs

@@ -0,0 +1,110 @@
+use std::sync::{Arc, Mutex};
+
+use http::Uri;
+use lru::LruCache;
+
+use crate::{
+    request_settings::AppliedRequestSettings,
+    transport::{CoreGrpcClient, PlatformGrpcClient},
+};
+
+/// ConnectionPool represents pool of connections to DAPI nodes.
+///
+/// It can be cloned and shared between threads.
+/// Cloning the pool will create a new reference to the same pool.
+#[derive(Debug, Clone)]
+pub struct ConnectionPool {
+    inner: Arc<Mutex<LruCache<String, PoolItem>>>,
+}
+
+impl ConnectionPool {
+    /// Create a new pool with a given capacity.
+    /// The pool will evict the least recently used item when the capacity is reached.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the capacity is zero.
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(LruCache::new(
+                capacity.try_into().expect("must be non-zero"),
+            ))),
+        }
+    }
+}
+
+impl Default for ConnectionPool {
+    fn default() -> Self {
+        Self::new(50)
+    }
+}
+
+impl ConnectionPool {
+    /// Get item from the pool for the given uri and settings.
+    pub fn get(&self, uri: &Uri, settings: Option<&AppliedRequestSettings>) -> Option<PoolItem> {
+        let key = format!("{}{:?}", uri, settings);
+        self.inner.lock().expect("must lock").get(&key).cloned()
+    }
+
+    /// Get value from cache or create it using provided closure.
+    /// If value is already in the cache, it will be returned.
+    /// If value is not in the cache, it will be created by calling `create()` and stored in the cache.
+    pub fn get_or_create(
+        &self,
+        uri: &Uri,
+        settings: Option<&AppliedRequestSettings>,
+        create: impl FnOnce() -> PoolItem,
+    ) -> PoolItem {
+        if let Some(cli) = self.get(uri, settings) {
+            return cli;
+        }
+
+        let cli = create();
+        self.put(uri, settings, cli.clone());
+        cli
+    }
+
+    /// Put item into the pool for the given uri and settings.
+    pub fn put(&self, uri: &Uri, settings: Option<&AppliedRequestSettings>, value: PoolItem) {
+        let key = format!("{}{:?}", uri, settings);
+        self.inner.lock().expect("must lock").put(key, value);
+    }
+}
+
+/// Item stored in the pool.
+///
+/// We use an enum as we need to represent two different types of clients.
+#[derive(Clone, Debug)]
+pub enum PoolItem {
+    Core(CoreGrpcClient),
+    Platform(PlatformGrpcClient),
+}
+
+impl From<PlatformGrpcClient> for PoolItem {
+    fn from(client: PlatformGrpcClient) -> Self {
+        Self::Platform(client)
+    }
+}
+impl From<CoreGrpcClient> for PoolItem {
+    fn from(client: CoreGrpcClient) -> Self {
+        Self::Core(client)
+    }
+}
+
+impl From<PoolItem> for PlatformGrpcClient {
+    fn from(client: PoolItem) -> Self {
+        match client {
+            PoolItem::Platform(client) => client,
+            _ => panic!("ClientType is not Platform: {:?}", client),
+        }
+    }
+}
+
+impl From<PoolItem> for CoreGrpcClient {
+    fn from(client: PoolItem) -> Self {
+        match client {
+            PoolItem::Core(client) => client,
+            _ => panic!("ClientType is not Core: {:?}", client),
+        }
+    }
+}

packages/rs-dapi-client/src/dapi_client.rs

@@ -8,6 +8,7 @@ use std::time::Duration;
 use tracing::Instrument;
 
 use crate::address_list::AddressListError;
+use crate::connection_pool::ConnectionPool;
 use crate::{
     transport::{TransportClient, TransportRequest},
     Address, AddressList, CanRetry, RequestSettings,
@@ -64,16 +65,21 @@ pub trait DapiRequestExecutor {
 pub struct DapiClient {
     address_list: RwLock<AddressList>,
     settings: RequestSettings,
+    pool: ConnectionPool,
     #[cfg(feature = "dump")]
     pub(crate) dump_dir: Option<std::path::PathBuf>,
 }
 
 impl DapiClient {
     /// Initialize new [DapiClient] and optionally override default settings.
     pub fn new(address_list: AddressList, settings: RequestSettings) -> Self {
+        // multiply by 3 as we need to store core and platform addresses, and we want some spare capacity just in case
+        let address_count = 3 * address_list.len();
+
         Self {
             address_list: RwLock::new(address_list),
             settings,
+            pool: ConnectionPool::new(address_count),
             #[cfg(feature = "dump")]
             dump_dir: None,
         }
@@ -153,9 +159,13 @@ impl DapiRequestExecutor for DapiClient {
                 // It stays wrapped in `Result` since we want to return
                 // `impl Future<Output = Result<...>`, not a `Result` itself.
                 let address = address_result?;
+                let pool = self.pool.clone();
 
-                let mut transport_client =
-                    R::Client::with_uri_and_settings(address.uri().clone(), &applied_settings);
+                let mut transport_client = R::Client::with_uri_and_settings(
+                    address.uri().clone(),
+                    &applied_settings,
+                    &pool,
+                );
 
                 let response = transport_request
                     .execute_transport(&mut transport_client, &applied_settings)

packages/rs-dapi-client/src/lib.rs

@@ -3,6 +3,7 @@
 #![deny(missing_docs)]
 
 mod address_list;
+mod connection_pool;
 mod dapi_client;
 #[cfg(feature = "dump")]
 pub mod dump;

packages/rs-dapi-client/src/transport.rs

@@ -2,6 +2,7 @@
 
 pub(crate) mod grpc;
 
+use crate::connection_pool::ConnectionPool;
 pub use crate::request_settings::AppliedRequestSettings;
 use crate::{CanRetry, RequestSettings};
 use dapi_grpc::mock::Mockable;
@@ -50,8 +51,12 @@ pub trait TransportClient: Send + Sized {
     type Error: CanRetry + Send + Debug;
 
     /// Build client using node's url.
-    fn with_uri(uri: Uri) -> Self;
+    fn with_uri(uri: Uri, pool: &ConnectionPool) -> Self;
 
     /// Build client using node's url and [AppliedRequestSettings].
-    fn with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Self;
+    fn with_uri_and_settings(
+        uri: Uri,
+        settings: &AppliedRequestSettings,
+        pool: &ConnectionPool,
+    ) -> Self;
 }

packages/rs-dapi-client/src/transport/grpc.rs

@@ -3,6 +3,7 @@
 use std::time::Duration;
 
 use super::{CanRetry, TransportClient, TransportRequest};
+use crate::connection_pool::ConnectionPool;
 use crate::{request_settings::AppliedRequestSettings, RequestSettings};
 use dapi_grpc::core::v0::core_client::CoreClient;
 use dapi_grpc::core::v0::{self as core_proto};
@@ -17,15 +18,13 @@ pub type PlatformGrpcClient = PlatformClient<Channel>;
 /// Core Client using gRPC transport.
 pub type CoreGrpcClient = CoreClient<Channel>;
 
-fn channel_with_uri(uri: Uri) -> Channel {
-    Channel::builder(uri).connect_lazy()
-}
-
-fn channel_with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Channel {
+fn create_channel(uri: Uri, settings: Option<&AppliedRequestSettings>) -> Channel {
     let mut builder = Channel::builder(uri);
 
-    if let Some(timeout) = settings.connect_timeout {
-        builder = builder.connect_timeout(timeout);
+    if let Some(settings) = settings {
+        if let Some(timeout) = settings.connect_timeout {
+            builder = builder.connect_timeout(timeout);
+        }
     }
 
     builder.connect_lazy()
@@ -34,24 +33,44 @@ fn channel_with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) ->
 impl TransportClient for PlatformGrpcClient {
     type Error = dapi_grpc::tonic::Status;
 
-    fn with_uri(uri: Uri) -> Self {
-        Self::new(channel_with_uri(uri))
+    fn with_uri(uri: Uri, pool: &ConnectionPool) -> Self {
+        pool.get_or_create(&uri, None, || {
+            Self::new(create_channel(uri.clone(), None)).into()
+        })
+        .into()
     }
 
-    fn with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Self {
-        Self::new(channel_with_uri_and_settings(uri, settings))
+    fn with_uri_and_settings(
+        uri: Uri,
+        settings: &AppliedRequestSettings,
+        pool: &ConnectionPool,
+    ) -> Self {
+        pool.get_or_create(&uri, Some(settings), || {
+            Self::new(create_channel(uri.clone(), Some(settings))).into()
+        })
+        .into()
     }
 }
 
 impl TransportClient for CoreGrpcClient {
     type Error = dapi_grpc::tonic::Status;
 
-    fn with_uri(uri: Uri) -> Self {
-        Self::new(channel_with_uri(uri))
+    fn with_uri(uri: Uri, pool: &ConnectionPool) -> Self {
+        pool.get_or_create(&uri, None, || {
+            Self::new(create_channel(uri.clone(), None)).into()
+        })
+        .into()
     }
 
-    fn with_uri_and_settings(uri: Uri, settings: &AppliedRequestSettings) -> Self {
-        Self::new(channel_with_uri_and_settings(uri, settings))
+    fn with_uri_and_settings(
+        uri: Uri,
+        settings: &AppliedRequestSettings,
+        pool: &ConnectionPool,
+    ) -> Self {
+        pool.get_or_create(&uri, Some(settings), || {
+            Self::new(create_channel(uri.clone(), Some(settings))).into()
+        })
+        .into()
     }
 }