group real messages a little

Cargo.lock

@@ -588,6 +588,7 @@ dependencies = [
  "gateway-requests",
  "gloo-timers",
  "humantime-serde",
+ "itertools",
  "log",
  "nonexhaustive-delayqueue",
  "nymsphinx",
@@ -2530,9 +2531,9 @@ dependencies = [
 
 [[package]]
 name = "itertools"
-version = "0.10.3"
+version = "0.10.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9a9d19fa1e79b6215ff29b9d6880b706147f16e9b1dbb1e4e5947b5b02bc5e3"
+checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473"
 dependencies = [
  "either",
 ]

clients/client-core/Cargo.toml

@@ -31,6 +31,7 @@ validator-client = { path = "../../common/client-libs/validator-client", default
 tap = "1.0.1"
 
 tokio = { version = "1.21.2", features = ["time", "macros"]}
+itertools = "0.10.5"
 
 [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen-futures]
 version = "0.4"
@@ -56,4 +57,4 @@ tempfile = "3.1.0"
 default = ["reply-surb"]
 wasm = ["gateway-client/wasm"]
 coconut = ["gateway-client/coconut", "gateway-requests/coconut"]
-reply-surb = ["sled"]
+reply-surb = ["sled"]

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

@@ -379,16 +379,26 @@ where
         if let Some(real_messages) = real_messages {
             // tells real message sender (with the poisson timer) to send this to the mix network
 
+            use itertools::Itertools;
+
             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];
+                let grouped: Vec<Vec<RealMessage>> = real_messages
+                    .into_iter()
+                    .chunks(10)
+                    .into_iter()
+                    .map(|c| c.collect())
+                    .collect();
+
+                for msg in grouped {
+                    //let msgs = vec![msg];
+                    //let msgs = msg.to_vec();
                     loop {
                         if self.real_message_sender.capacity() > 2 {
-                            if self.real_message_sender.send(msgs).await.is_err() {
+                            if self.real_message_sender.send(msg).await.is_err() {
                                 panic!();
                             }
                             break;

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

@@ -19,7 +19,8 @@ mod inbound;
 mod outbound;
 
 // TODO: make this configurable
-const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
+//const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(60 * 10);
 
 #[derive(Debug)]
 pub struct ProxyMessage {