prio

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

@@ -15,6 +15,7 @@ use nymsphinx::preparer::MessagePreparer;
 use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
 use rand::{CryptoRng, Rng};
 use std::sync::Arc;
+use std::time::Duration;
 
 #[cfg(feature = "reply-surb")]
 use crate::client::reply_key_storage::ReplyKeyStorage;
@@ -186,9 +187,9 @@ where
         // there's no point in trying to send nothing
         if let Some(real_messages) = real_messages {
             // tells real message sender (with the poisson timer) to send this to the mix network
-            self.real_message_sender
-                .unbounded_send(real_messages)
-                .unwrap();
+            if self.real_message_sender.send(real_messages).await.is_err() {
+                panic!();
+            }
         }
     }
 
@@ -377,9 +378,30 @@ where
         // there's no point in trying to send nothing
         if let Some(real_messages) = real_messages {
             // tells real message sender (with the poisson timer) to send this to the mix network
-            self.real_message_sender
-                .unbounded_send(real_messages)
-                .unwrap();
+
+            log::info!("chunked into {} messages", real_messages.len());
+            log::info!("current capacity: {}", self.real_message_sender.capacity());
+            if real_messages.len() > 10 {
+                // only send if there is nothing in queue
+                log::info!("sending large message(s)");
+                for msg in real_messages {
+                    let msgs = vec![msg];
+                    loop {
+                        if self.real_message_sender.capacity() > 2 {
+                            if self.real_message_sender.send(msgs).await.is_err() {
+                                panic!();
+                            }
+                            break;
+                        }
+                        tokio::time::sleep(Duration::from_millis(50)).await;
+                    }
+                }
+                log::error!("finished sending large message(s)");
+            } else {
+                if self.real_message_sender.send(real_messages).await.is_err() {
+                    panic!();
+                }
+            }
         }
     }
 }

clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs

@@ -112,12 +112,17 @@ where
             .unwrap();
 
         // send to `OutQueueControl` to eventually send to the mix network
-        self.real_message_sender
-            .unbounded_send(vec![RealMessage::new(
+        if self
+            .real_message_sender
+            .send(vec![RealMessage::new(
                 prepared_fragment.mix_packet,
                 frag_id,
             )])
-            .unwrap();
+            .await
+            .is_err()
+        {
+            panic!();
+        }
     }
 
     #[cfg(not(target_arch = "wasm32"))]

clients/client-core/src/client/real_messages_control/mod.rs

@@ -123,7 +123,7 @@ impl RealMessagesController<OsRng> {
     ) -> Self {
         let rng = OsRng;
 
-        let (real_message_sender, real_message_receiver) = mpsc::unbounded();
+        let (real_message_sender, real_message_receiver) = tokio::sync::mpsc::channel(3);
         let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded();
 
         let ack_controller_connectors = AcknowledgementControllerConnectors::new(

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

@@ -242,8 +242,8 @@ impl RealMessage {
 
 // messages are already prepared, etc. the real point of it is to forward it to mix_traffic
 // after sufficient delay
-pub type BatchRealMessageSender = mpsc::UnboundedSender<Vec<RealMessage>>;
-type BatchRealMessageReceiver = mpsc::UnboundedReceiver<Vec<RealMessage>>;
+pub type BatchRealMessageSender = tokio::sync::mpsc::Sender<Vec<RealMessage>>;
+type BatchRealMessageReceiver = tokio::sync::mpsc::Receiver<Vec<RealMessage>>;
 
 pub(crate) enum StreamMessage {
     Cover,
@@ -425,7 +425,7 @@ where
             }
 
             // decide what kind of message to send
-            match Pin::new(&mut self.real_receiver).poll_next(cx) {
+            match Pin::new(&mut self.real_receiver).poll_recv(cx) {
                 // in the case our real message channel stream was closed, we should also indicate we are closed
                 // (and whoever is using the stream should panic)
                 Poll::Ready(None) => Poll::Ready(None),
@@ -472,7 +472,7 @@ where
             return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available))));
         }
 
-        match Pin::new(&mut self.real_receiver).poll_next(cx) {
+        match Pin::new(&mut self.real_receiver).poll_recv(cx) {
             // in the case our real message channel stream was closed, we should also indicate we are closed
             // (and whoever is using the stream should panic)
             Poll::Ready(None) => Poll::Ready(None),

common/socks5/proxy-helpers/src/proxy_runner/inbound.rs

@@ -70,8 +70,10 @@ where
 
     // WIP(JON): here we do the chunking, and send to real_message_sender instead
     if let Some(chunker) = msg_chunker {
+        log::info!("({connection_id}): chunking and sending");
         let msg = adapter_fn(connection_id, ordered_msg, is_finished);
         chunker.chunk(msg).await;
+        log::info!("({connection_id}): chunking and sending done");
     } else {
         mix_sender
             .unbounded_send(adapter_fn(connection_id, ordered_msg, is_finished))