calling for shutdown from the MixTrafficController

Author: simonwicky

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

@@ -801,7 +801,7 @@ where
             event_tx,
         );
 
-        let mix_tx = mix_traffic_controller.mix_rx();
+        let mix_tx = mix_traffic_controller.mix_tx();
         let client_tx = mix_traffic_controller.client_tx();
 
         shutdown_tracker.try_spawn_named(

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

@@ -87,7 +87,7 @@ impl MixTrafficController {
         self.client_tx.clone()
     }
 
-    pub fn mix_rx(&self) -> BatchMixMessageSender {
+    pub fn mix_tx(&self) -> BatchMixMessageSender {
         self.mix_tx.clone()
     }
 
@@ -159,6 +159,11 @@ impl MixTrafficController {
                             // Do we need to handle the embedded mixnet client case
                             // separately?
                             self.event_tx.send(MixnetClientEvent::Traffic(MixTrafficEvent::FailedSendingSphinx));
+                            // IMO it shouldn't be signalled from there but it is how it is
+                            // TODO : report the failure upwards and shutdown from upwards
+                            // Gateway is dead, we have to shut down currently
+                            error!("Signalling shutdown from the MixTrafficController");
+                            self.shutdown_token.cancel();
                             break;
                         }
                     }

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

@@ -598,20 +598,15 @@ where
                         tracing::trace!("OutQueueControl: Received shutdown");
                         break;
                     }
-                _ = status_timer.tick() => {
-                    self.log_status();
-                }
-                next_message = self.next() => if let Some(next_message) = next_message {
-                    // Check if mix_tx channel is closed BEFORE processing message
-                    if self.mix_tx.is_closed() {
-                        tracing::error!("OutQueueControl: mix_tx channel closed, stopping traffic stream");
+                    _ = status_timer.tick() => {
+                        self.log_status();
+                    }
+                    next_message = self.next() => if let Some(next_message) = next_message {
+                        self.on_message(next_message).await;
+                    } else {
+                        tracing::trace!("OutQueueControl: Stopping since channel closed");
                         break;
                     }
-                    self.on_message(next_message).await;
-                } else {
-                    tracing::trace!("OutQueueControl: Stopping since channel closed");
-                    break;
-                }
                 }
             }
         }
@@ -626,11 +621,6 @@ where
                         break;
                     }
                     next_message = self.next() => if let Some(next_message) = next_message {
-                        // Check if mix_tx channel is closed BEFORE processing message
-                        if self.mix_tx.is_closed() {
-                            tracing::error!("OutQueueControl: mix_tx channel closed, stopping traffic stream");
-                            break;
-                        }
                         self.on_message(next_message).await;
                     } else {
                         tracing::trace!("OutQueueControl: Stopping since channel closed");