hookup heartbeat in connection recovery process

Author: Keruspe

src/channels.rs

@@ -280,8 +280,8 @@ impl Channels {
     }
 
     pub(crate) fn init_connection_recovery(&self, error: Error) {
+        self.heartbeat.reset();
         self.connection_status.set_reconnecting();
-        // FIXME: reset heartbeat
         self.lock_inner()
             .channels
             .values()
@@ -292,7 +292,7 @@ impl Channels {
     }
 
     pub(crate) async fn start_recovery(&self) -> Result<()> {
-        // FIXME: reopen connection and restart heartbeat
+        // FIXME: reopen connection
 
         let channels = self
             .lock_inner()
@@ -307,6 +307,8 @@ impl Channels {
             channel.start_recovery().await?;
         }
 
+        self.start_heartbeat();
+
         Ok(())
     }
 

src/heartbeat.rs

@@ -72,6 +72,11 @@ impl Heartbeat {
         self.lock_inner().timeout = None;
     }
 
+    pub(crate) fn reset(&self) {
+        self.killswitch.reset();
+        self.lock_inner().reset();
+    }
+
     fn lock_inner(&self) -> MutexGuard<'_, Inner> {
         self.inner.lock().unwrap_or_else(|e| e.into_inner())
     }
@@ -130,4 +135,10 @@ impl Inner {
     fn update_last_read(&mut self) {
         self.last_read = Instant::now();
     }
+
+    fn reset(&mut self) {
+        self.update_last_read();
+        self.update_last_write();
+        self.timeout = None;
+    }
 }

src/killswitch.rs

@@ -14,4 +14,8 @@ impl KillSwitch {
     pub(crate) fn killed(&self) -> bool {
         self.0.load(Ordering::SeqCst)
     }
+
+    pub(crate) fn reset(&self) {
+        self.0.store(false, Ordering::SeqCst)
+    }
 }