Remove all recv_with_delay and add shutdown condition to loops in client-core

Inside client-core we want to prepare the ground for moving a behaviour
close to what we have in the vpn client.

Remove all the recv_with_delay since we want to just stop

Add shutdown condition to all select loops to guard against the shutdown
listener being polled inside the select blocks.

common/client-core/src/client/mix_traffic/mod.rs

@@ -113,7 +113,7 @@ impl MixTrafficController {
         spawn_future(async move {
             debug!("Started MixTrafficController with graceful shutdown support");
 
-            loop {
+            while !shutdown.is_shutdown() {
                 tokio::select! {
                     mix_packets = self.mix_rx.recv() => match mix_packets {
                         Some(mix_packets) => {
@@ -146,7 +146,7 @@ impl MixTrafficController {
                             log::trace!("MixTrafficController, client request channel closed");
                         }
                     },
-                    _ = shutdown.recv_with_delay() => {
+                    _ = shutdown.recv() => {
                         log::trace!("MixTrafficController: Received shutdown");
                         break;
                     }

common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs

@@ -88,7 +88,7 @@ impl AcknowledgementListener {
                         break;
                     }
                 },
-                _ = shutdown.recv_with_delay() => {
+                _ = shutdown.recv() => {
                     log::trace!("AcknowledgementListener: Received shutdown");
                 }
             }

common/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs

@@ -268,7 +268,7 @@ impl ActionController {
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
         debug!("Started ActionController with graceful shutdown support");
 
-        loop {
+        while !shutdown.is_shutdown() {
             tokio::select! {
                 action = self.incoming_actions.next() => match action {
                     Some(action) => self.process_action(action),
@@ -286,7 +286,7 @@ impl ActionController {
                         break;
                     }
                 },
-                _ = shutdown.recv_with_delay() => {
+                _ = shutdown.recv() => {
                     log::trace!("ActionController: Received shutdown");
                     break;
                 }

common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs

@@ -178,7 +178,7 @@ where
                         break;
                     }
                 },
-                _ = shutdown.recv_with_delay() => {
+                _ = shutdown.recv() => {
                     log::trace!("InputMessageListener: Received shutdown");
                 }
             }

common/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs

@@ -40,7 +40,7 @@ impl SentNotificationListener {
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
         debug!("Started SentNotificationListener with graceful shutdown support");
 
-        loop {
+        while !shutdown.is_shutdown() {
             tokio::select! {
                 frag_id = self.sent_notifier.next() => match frag_id {
                     Some(frag_id) => {
@@ -51,7 +51,7 @@ impl SentNotificationListener {
                         break;
                     }
                 },
-                _ = shutdown.recv_with_delay() => {
+                _ = shutdown.recv() => {
                     log::trace!("SentNotificationListener: Received shutdown");
                     break;
                 }

common/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -542,7 +542,7 @@ where
         {
             let mut status_timer = tokio::time::interval(Duration::from_secs(5));
 
-            loop {
+            while !shutdown.is_shutdown() {
                 tokio::select! {
                     biased;
                     _ = shutdown.recv() => {

common/client-core/src/client/received_buffer.rs

@@ -428,7 +428,7 @@ impl<R: MessageReceiver> RequestReceiver<R> {
         while !shutdown.is_shutdown() {
             tokio::select! {
                 biased;
-                _ = shutdown.recv_with_delay() => {
+                _ = shutdown.recv() => {
                     log::trace!("RequestReceiver: Received shutdown");
                 }
                 request = self.query_receiver.next() => {
@@ -441,7 +441,7 @@ impl<R: MessageReceiver> RequestReceiver<R> {
                 },
             }
         }
-        shutdown.recv_timeout().await;
+        shutdown.recv().await;
         log::debug!("RequestReceiver: Exiting");
     }
 }

common/client-core/src/client/replies/reply_controller/mod.rs

@@ -860,7 +860,7 @@ where
         while !shutdown.is_shutdown() {
             tokio::select! {
                 biased;
-                _ = shutdown.recv_with_delay() => {
+                _ = shutdown.recv() => {
                     log::trace!("ReplyController: Received shutdown");
                 },
                 req = self.request_receiver.next() => match req {

common/client-core/src/client/statistics_control.rs

@@ -121,7 +121,7 @@ impl StatisticsControl {
         let mut snapshot_interval =
             gloo_timers::future::IntervalStream::new(SNAPSHOT_INTERVAL.as_millis() as u32);
 
-        loop {
+        while !task_client.is_shutdown() {
             tokio::select! {
                 biased;
                 _ = task_client.recv() => {