WIP

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

@@ -92,7 +92,8 @@ where
             // while we were messing with topology, wrapping data in sphinx, etc. we actually received
             // this ack after all! no need to retransmit then
             debug!("We received an ack JUST as we were about to retransmit [2]");
-            return;
+            panic!("will we reach this?");
+            //return;
         }
         // we no longer need the reference - let's drop it so that if somehow `UpdateTimer` action
         // reached the controller before this function terminated, the controller would not panic.

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

@@ -94,7 +94,7 @@ impl Config {
 
 pub struct RealMessagesController<R>
 where
-    R: CryptoRng + Rng,
+    R: CryptoRng + Rng + Clone,
 {
     out_queue_control: OutQueueControl<R>,
     ack_control: AcknowledgementController<R>,

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

@@ -15,6 +15,7 @@ use nymsphinx::cover::generate_loop_cover_packet;
 use nymsphinx::forwarding::packet::MixPacket;
 use nymsphinx::params::PacketSize;
 use nymsphinx::utils::sample_poisson_duration;
+use rand::rngs::OsRng;
 use rand::{CryptoRng, Rng};
 use std::collections::VecDeque;
 use std::pin::Pin;
@@ -182,7 +183,7 @@ impl SendingDelayController {
 
 pub(crate) struct OutQueueControl<R>
 where
-    R: CryptoRng + Rng,
+    R: CryptoRng + Rng + Clone,
 {
     /// Configurable parameters of the `ActionController`
     config: Config,
@@ -252,7 +253,7 @@ pub(crate) enum StreamMessage {
 
 impl<R> OutQueueControl<R>
 where
-    R: CryptoRng + Rng + Unpin,
+    R: CryptoRng + Rng + Unpin + Clone,
 {
     // at this point I'm not entirely sure how to deal with this warning without
     // some considerable refactoring
@@ -535,11 +536,23 @@ where
             self.on_message(next_message).await;
         }
     }
+
+    fn poisson_stream(&mut self) -> impl futures::Stream<Item = Duration> {
+        //let avg_delay = self.current_average_message_sending_delay();
+        futures::stream::unfold((), |_| async {
+            let mut rng = OsRng;
+            //let next_poisson_delay = sample_poisson_duration(&mut rng, avg_delay);
+            //time::sleep(Duration::from(next_poisson_delay)).await;
+            let avg_delay = Duration::from_millis(10);
+            time::sleep(avg_delay).await;
+            Some((avg_delay, ()))
+        })
+    }
 }
 
 impl<R> Stream for OutQueueControl<R>
 where
-    R: CryptoRng + Rng + Unpin,
+    R: CryptoRng + Rng + Unpin + Clone,
 {
     type Item = StreamMessage;