Reconnecting sockets patch. Fix clippy lints.

Author: ThatOneToast

Cargo.lock

@@ -25,3 +25,4 @@ tokio = { version = "=1", features = ["full"] }
 uuid = { version = "=1", features = ["v4"] }
 
 tcrypt = { git = "https://github.com/thatonetoast/tcrypt.git", tag = "0.1.1-1" }
+scopeguard = "1.2.0"

Cargo.toml

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use tokio::sync::RwLock;
 
-use crate::packet;
+use crate::{errors::Error, packet};
 
 use super::client::AsyncClient;
 
@@ -22,4 +22,24 @@ impl<P: packet::Packet> AsyncClientRef<P> {
     pub async fn read(&self) -> tokio::sync::RwLockReadGuard<'_, AsyncClient<P>> {
         self.0.read().await
     }
+
+    pub async fn signal_reconnect(&self) -> Result<(), Error> {
+        let client = self.0.read().await;
+
+        if let Some(tx) = &client.keepalive_reconnect_tx {
+            // Clone the channel to avoid holding the read lock during send
+            let tx_clone = tx.clone();
+            drop(client);
+
+            // Send the reconnect signal
+            match tx_clone.send(()).await {
+                Ok(()) => Ok(()),
+                Err(_) => Err(Error::Other("Failed to signal reconnection".to_string())),
+            }
+        } else {
+            Err(Error::Other(
+                "Keepalive reconnection channel not available".to_string(),
+            ))
+        }
+    }
 }

src/asynch/client.rs

@@ -317,6 +317,7 @@ where
     /// # Panics
     ///
     /// * Panics if the specified pool doesn't exist
+    #[allow(clippy::significant_drop_tightening)]
     pub async fn add_socket_to_pool(&mut self, pool_name: &str, socket: &TSocket<S>) {
         let mut pool = self.pools.write().await;
         let pool = pool.get_mut(pool_name).expect("Unknown Pool");
@@ -401,6 +402,7 @@ where
     /// # Returns
     ///
     /// * `Result<Option<Encryptor>, Error>` - The encryption configuration or an error
+    #[allow(clippy::significant_drop_in_scrutinee)]
     async fn handle_authentication(
         &mut self,
         tsocket: &mut TSocket<S>,
@@ -507,6 +509,7 @@ where
     ///     listener.broadcast(packet).await.expect("Broadcast failed");
     /// }
     /// ```
+    #[allow(clippy::significant_drop_in_scrutinee)]
     pub async fn broadcast(&self, packet: P) -> Result<(), Error> {
         let pool = self.keep_alive_pool.clone().sockets;
         for socket in pool.write().await.iter_mut() {

src/asynch/client_ext.rs

@@ -22,7 +22,6 @@ pub mod errors;
 pub mod macros;
 pub mod packet;
 pub mod phantom;
-pub mod reconnect;
 pub mod resources;
 pub mod session;
 

src/asynch/listener.rs

@@ -95,6 +95,7 @@ impl ImplSession for TestSession {
 
 // Define test resource
 #[derive(Debug, Clone)]
+#[allow(dead_code)]
 struct TestResource {
     data: Vec<String>,
 }
@@ -530,20 +531,8 @@ async fn test_max_retries_exceeded() {
     // For this test, we still need a client struct to configure reconnection parameters
     // but the initial connection attempt will fail
 
-    // Create reconnection config with shorter timeouts for testing
-    let config = ReconnectionConfig {
-        endpoints: vec![],
-        auto_reconnect: true,
-        max_attempts: Some(2),
-        initial_retry_delay: 0.1,
-        max_retry_delay: 0.2,
-        backoff_factor: 1.5,
-        jitter: 0.0,
-        reinitialize: true,
-    };
 
-    // Use the port simulator approach to test the behavior
-    let reconnect_config = config.clone();
+
 
     // This test is mainly to ensure the client handles max retry limits gracefully
     // We'll make an attempt to connect to a non-existent server
@@ -564,3 +553,118 @@ async fn test_max_retries_exceeded() {
     // The actual logic for max retries is handled within the client.rs implementation
     // and is exercised by the other tests in a more realistic way
 }
+
+// Test 6: Reconnection after server downtime
+#[tokio::test]
+async fn test_reconnection_after_downtime() {
+    let port = 9096;
+
+    // Start a server
+    let (server_stop_tx, server_stop_rx) = oneshot::channel();
+    start_test_server(port, server_stop_rx).await;
+
+    // Give the server time to start
+    sleep(Duration::from_millis(500)).await;
+
+    // Create a client with reconnection enabled
+    let client_result = AsyncClient::<TestPacket>::new("127.0.0.1", port).await;
+    if client_result.is_err() {
+        println!("Skipping test_reconnection_after_downtime as we can't create initial client");
+        let _ = server_stop_tx.send(());
+        return;
+    }
+
+    let mut client = client_result
+        .unwrap()
+        .with_reconnection(ReconnectionConfig {
+            endpoints: vec![],
+            auto_reconnect: true,
+            max_attempts: Some(10),
+            initial_retry_delay: 0.1, // Fast retries for testing
+            max_retry_delay: 1.0,
+            backoff_factor: 1.5,
+            jitter: 0.1,
+            reinitialize: true,
+        });
+
+    // Initialize the connection
+    client.finalize().await;
+
+    // Establish a session by sending an initial request
+    let initial_packet = TestPacket::ok();
+    let initial_response = match client.send_recv(initial_packet).await {
+        Ok(response) => response,
+        Err(e) => {
+            println!("Skipping test as we could not establish initial session: {:?}", e);
+            let _ = server_stop_tx.send(());
+            return;
+        }
+    };
+
+    // Verify we have a session
+    let initial_session_id = initial_response.body().session_id.clone();
+    assert!(initial_session_id.is_some(), "No session ID in initial response");
+    println!("Initial session ID: {:?}", initial_session_id);
+
+    // Stop the server
+    server_stop_tx.send(()).unwrap();
+    println!("Server stopped, waiting for 5 seconds...");
+
+    // Wait for 5 seconds to simulate extended downtime
+    sleep(Duration::from_secs(5)).await;
+
+    // Start a new server
+    let (new_server_stop_tx, new_server_stop_rx) = oneshot::channel();
+    let new_server_handle = start_test_server(port, new_server_stop_rx).await;
+
+    // Give the new server time to start
+    sleep(Duration::from_millis(500)).await;
+    println!("New server started");
+
+    // Prepare for reconnection attempts
+    let reconnect_start = Instant::now();
+    let mut reconnected = false;
+    let max_reconnect_time = Duration::from_secs(10);
+
+    // Send a packet which should trigger reconnection
+    while reconnect_start.elapsed() < max_reconnect_time {
+        let test_packet = TestPacket {
+            header: "TEST".to_string(),
+            body: PacketBody::default(),
+            data: Some("reconnect after downtime".to_string()),
+        };
+
+        match client.send_recv(test_packet).await {
+            Ok(response) => {
+                // Successfully reconnected
+                assert_eq!(response.header(), "OK");
+                println!("Successfully reconnected after 5 seconds of downtime");
+
+                // Verify the response contains data
+                if let Some(data) = &response.data {
+                    assert!(data.contains("reconnect after downtime"));
+                }
+
+                // Check if we got a new session
+                let new_session_id = response.body().session_id;
+                println!("New session ID after reconnection: {:?}", new_session_id);
+
+                reconnected = true;
+                break;
+            }
+            Err(e) => {
+                println!("Reconnection attempt failed: {:?}, retrying...", e);
+                sleep(Duration::from_millis(500)).await;
+            }
+        }
+    }
+
+    // Assert that we were able to reconnect
+    assert!(reconnected, "Failed to reconnect after server downtime");
+
+    // Clean up
+    new_server_stop_tx.send(()).unwrap();
+    tokio::time::timeout(Duration::from_secs(2), new_server_handle)
+        .await
+        .ok();
+}