- Fix silly bug in pool creation

- Remove unused function
- Fix the way a connection is created by passing the same scheme as the main one

Author: madchicken

lib/src/connection.rs

@@ -455,7 +455,7 @@ impl NeoUrl {
         Ok(Self(url))
     }
 
-    fn scheme(&self) -> &str {
+    pub(crate) fn scheme(&self) -> &str {
         self.0.scheme()
     }
 

lib/src/graph.rs

@@ -74,10 +74,7 @@ impl Graph {
                 &config.tls_config,
             )?;
             if matches!(info.routing, Routing::Yes(_)) {
-                let pool = Routed(
-                    RoutedConnectionManager::new(&config, Arc::new(RoundRobinStrategy::default()))
-                        .await?,
-                );
+                let pool = Routed(RoutedConnectionManager::new(&config).await?);
                 Ok(Graph {
                     config: config.into_live_config(),
                     pool,

lib/src/routing/connection_registry.rs

@@ -46,9 +46,8 @@ pub(crate) struct ConnectionRegistry {
 }
 
 impl ConnectionRegistry {
-    pub(crate) async fn new(config: &Config) -> Result<Self, Error> {
-        let connections = Self::build_registry(config, &[]).await?;
-        Ok(ConnectionRegistry {
+    pub(crate) fn new(config: &Config) -> Self {
+        ConnectionRegistry {
             config: config.clone(),
             creation_time: Arc::new(Mutex::new(
                 std::time::SystemTime::now()
@@ -57,20 +56,8 @@ impl ConnectionRegistry {
                     .as_secs(),
             )),
             ttl: Arc::new(AtomicU64::new(0)),
-            connections,
-        })
-    }
-
-    async fn build_registry(config: &Config, servers: &[BoltServer]) -> Result<Registry, Error> {
-        let registry = DashMap::new();
-        for server in servers.iter() {
-            let server_config = Config {
-                uri: format!("{}:{}", server.address, server.port), // build a config for each server in the routing table
-                ..config.clone() // but keep the information about tls and other settings
-            };
-            registry.insert(server.clone(), create_pool(&server_config).await?);
+            connections: DashMap::new(),
         }
-        Ok(registry)
     }
 
     pub(crate) async fn update_if_expired<F, R>(&self, f: F) -> Result<(), Error>
@@ -90,11 +77,20 @@ impl ConnectionRegistry {
                 debug!("Routing table refreshed: {:?}", routing_table);
                 let registry = &self.connections;
                 let servers = routing_table.resolve();
+                let url = NeoUrl::parse(self.config.uri.as_str())?;
+
                 for server in servers.iter() {
                     if registry.contains_key(server) {
                         continue;
                     }
-                    registry.insert(server.clone(), create_pool(&self.config).await?);
+                    registry.insert(
+                        server.clone(),
+                        create_pool(&Config {
+                            uri: format!("{}://{}:{}", url.scheme(), server.address, server.port),
+                            ..self.config.clone()
+                        })
+                        .await?,
+                    );
                 }
                 registry.retain(|k, _| servers.contains(k));
                 let _ = self
@@ -185,7 +181,7 @@ mod tests {
             fetch_size: 0,
             tls_config: ConnectionTLSConfig::None,
         };
-        let registry = ConnectionRegistry::new(&config).await.unwrap();
+        let registry = ConnectionRegistry::new(&config);
         registry
             .update_if_expired(|| async { Ok(cluster_routing_table) })
             .await

lib/src/routing/routed_connection_manager.rs

@@ -2,6 +2,7 @@ use crate::connection::{Connection, ConnectionInfo};
 use crate::pool::ManagedConnection;
 use crate::routing::connection_registry::{BoltServer, ConnectionRegistry};
 use crate::routing::load_balancing::LoadBalancingStrategy;
+use crate::routing::RoundRobinStrategy;
 #[cfg(feature = "unstable-bolt-protocol-impl-v2")]
 use crate::routing::{RouteBuilder, RoutingTable};
 use crate::{Config, Error, Operation};
@@ -14,19 +15,16 @@ use std::time::Duration;
 #[derive(Clone)]
 pub struct RoutedConnectionManager {
     load_balancing_strategy: Arc<dyn LoadBalancingStrategy>,
-    registry: Arc<ConnectionRegistry>,
+    connection_registry: Arc<ConnectionRegistry>,
     #[allow(dead_code)]
     bookmarks: Arc<Mutex<Vec<String>>>,
     backoff: Arc<ExponentialBackoff>,
     config: Config,
 }
 
 impl RoutedConnectionManager {
-    pub async fn new(
-        config: &Config,
-        load_balancing_strategy: Arc<dyn LoadBalancingStrategy>,
-    ) -> Result<Self, Error> {
-        let registry = Arc::new(ConnectionRegistry::new(config).await?);
+    pub async fn new(config: &Config) -> Result<Self, Error> {
+        let registry = Arc::new(ConnectionRegistry::new(config));
         let backoff = Arc::new(
             ExponentialBackoffBuilder::new()
                 .with_initial_interval(Duration::from_millis(1))
@@ -37,8 +35,8 @@ impl RoutedConnectionManager {
         );
 
         Ok(RoutedConnectionManager {
-            load_balancing_strategy,
-            registry,
+            load_balancing_strategy: Arc::new(RoundRobinStrategy::default()),
+            connection_registry: registry,
             bookmarks: Arc::new(Mutex::new(vec![])),
             backoff,
             config: config.clone(),
@@ -70,7 +68,7 @@ impl RoutedConnectionManager {
     ) -> Result<ManagedConnection, Error> {
         // We probably need to do this in a more efficient way, since this will block the request of a connection
         // while we refresh the routing table. We should probably have a separate thread that refreshes the routing
-        self.registry
+        self.connection_registry
             .update_if_expired(|| self.refresh_routing_table())
             .await?;
 
@@ -80,15 +78,15 @@ impl RoutedConnectionManager {
             _ => self.select_reader(),
         } {
             debug!("requesting connection for server: {:?}", server);
-            if let Some(pool) = self.registry.get_pool(&server) {
+            if let Some(pool) = self.connection_registry.get_pool(&server) {
                 match pool.get().await {
                     Ok(connection) => return Ok(connection),
                     Err(e) => {
                         error!(
                             "Failed to get connection from pool for server `{}`: {}",
                             server.address, e
                         );
-                        self.registry.mark_unavailable(&server);
+                        self.connection_registry.mark_unavailable(&server);
                         continue;
                     }
                 }
@@ -115,11 +113,11 @@ impl RoutedConnectionManager {
 
     fn select_reader(&self) -> Option<BoltServer> {
         self.load_balancing_strategy
-            .select_reader(&self.registry.servers())
+            .select_reader(&self.connection_registry.servers())
     }
 
     fn select_writer(&self) -> Option<BoltServer> {
         self.load_balancing_strategy
-            .select_writer(&self.registry.servers())
+            .select_writer(&self.connection_registry.servers())
     }
 }

lib/tests/use_default_db.rs

@@ -40,8 +40,8 @@ async fn use_default_db() {
         if default_db != dbname {
             eprintln!(
                 concat!(
-                "Skipping test: The test must run against a testcontainer ",
-                "or have `{}` configured as the default database"
+                    "Skipping test: The test must run against a testcontainer ",
+                    "or have `{}` configured as the default database"
                 ),
                 dbname
             );
@@ -63,16 +63,15 @@ async fn use_default_db() {
         if default_db != dbname {
             eprintln!(
                 concat!(
-                "Skipping test: The test must run against a testcontainer ",
-                "or have `{}` configured as the default database"
+                    "Skipping test: The test must run against a testcontainer ",
+                    "or have `{}` configured as the default database"
                 ),
                 dbname
             );
             return;
         }
     }
 
-
     let id = uuid::Uuid::new_v4();
     graph
         .run(query("CREATE (:Node { uuid: $uuid })").param("uuid", id.to_string()))
@@ -103,7 +102,7 @@ async fn use_default_db() {
             .execute_on(
                 dbname.as_str(),
                 query("MATCH (n:Node {uuid: $uuid}) RETURN count(n) AS result")
-                    .param("uuid", id.to_string())
+                    .param("uuid", id.to_string()),
             )
             .await
             .unwrap()