(iggy) branch master updated: fix(rust): add heartbeat handle and drop for rust sdk (#3852)

[ri...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

slbotbm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git


The following commit(s) were added to refs/heads/master by this push:
     new 0102c0f81 fix(rust): add heartbeat handle and drop for rust sdk (#3852)
0102c0f81 is described below

commit 0102c0f81cea7eea50b78a5a4d35ebb1735d6a3d
Author: Rimuksh Kansal <ni...@gmail.com>
AuthorDate: Mon Aug 10 20:03:52 2026 +0900

    fix(rust): add heartbeat handle and drop for rust sdk (#3852)
    
    ## Which issue does this PR address?
    
    Relates to
    https://github.com/apache/iggy/pull/3733#discussion_r3728698168
    
    ## Rationale
    
    The heartbeat handle was being discarded and on shutdown, not being
    stopped on calling `shutdown`.
    
    ## What changed?
    
    Adds a heartbeat_handler that is saved when `connect()` is called. In
    case connect is called repeatedly, the code uses the handler, and drops
    it when the client is destroyed.
    
    ## Local Execution
    
    - Passed
    - Pre-commit hooks ran
    
    
    ## AI Usage
    
    If AI tools were used, please answer:
    1. Which tools? codex
    2. Scope of usage? code generation
    3. How did you verify the generated code works correctly? tested it
    4. Can you explain every line of the code if asked? yes
    
    Co-authored-by: Piotr Gankiewicz <pi...@gmail.com>
    Co-authored-by: THE Patryk <ph...@gmail.com>
---
 core/sdk/src/clients/client.rs | 35 ++++++++++++++++++++++++++++++++---
 1 file changed, 32 insertions(+), 3 deletions(-)

diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index dd7f91fd9..e91a601a5 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -39,8 +39,9 @@ use iggy_common::locking::{IggyRwLock, IggyRwLockFn};
 use iggy_common::{BinaryTransport, Client, HttpMethod, SystemClient};
 use iggy_common::{ConnectionStringUtils, DiagnosticEvent, Partitioner, TransportProtocol};
 use std::fmt::Debug;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use tokio::spawn;
+use tokio::task::JoinHandle;
 use tokio::time::sleep;
 use tracing::log::warn;
 use tracing::{debug, error, info};
@@ -64,6 +65,7 @@ pub struct IggyClient {
     pub(crate) client: IggyRwLock<ClientWrapper>,
     partitioner: Option<Arc<dyn Partitioner>>,
     pub(crate) encryptor: Option<Arc<EncryptorKind>>,
+    heartbeat_handle: Mutex<Option<JoinHandle<()>>>,
 }
 
 impl Default for IggyClient {
@@ -92,6 +94,7 @@ impl IggyClient {
             client,
             partitioner: None,
             encryptor: None,
+            heartbeat_handle: Mutex::new(None),
         }
     }
 
@@ -131,6 +134,7 @@ impl IggyClient {
             client,
             partitioner,
             encryptor,
+            heartbeat_handle: Mutex::new(None),
         }
     }
 
@@ -236,6 +240,19 @@ impl IggyClient {
     }
 }
 
+impl Drop for IggyClient {
+    fn drop(&mut self) {
+        let heartbeat_handle = self
+            .heartbeat_handle
+            .get_mut()
+            .unwrap_or_else(|error| error.into_inner())
+            .take();
+        if let Some(handle) = heartbeat_handle {
+            handle.abort();
+        }
+    }
+}
+
 #[async_trait]
 impl Client for IggyClient {
     async fn connect(&self) -> Result<(), IggyError> {
@@ -246,8 +263,20 @@ impl Client for IggyClient {
             heartbeat_interval = client.heartbeat_interval().await;
         }
 
+        let mut heartbeat_handle = self
+            .heartbeat_handle
+            .lock()
+            .unwrap_or_else(|error| error.into_inner());
+        if heartbeat_handle
+            .as_ref()
+            .is_some_and(|handle| !handle.is_finished())
+        {
+            return Ok(());
+        }
+
+        drop(heartbeat_handle.take());
         let client = self.client.clone();
-        spawn(async move {
+        *heartbeat_handle = Some(spawn(async move {
             loop {
                 debug!("Sending the heartbeat...");
                 if let Err(error) = client.read().await.ping().await {
@@ -268,7 +297,7 @@ impl Client for IggyClient {
                 }
                 sleep(heartbeat_interval.get_duration()).await
             }
-        });
+        }));
         Ok(())
     }