Implementing configuration point shipper_throttle_packet_delay_ms

config/src/config.rs

@@ -36,6 +36,7 @@ pub struct Config {
     //0 = OFF (don't run GC)
     //Default is 10000 (10 seconds)
     pub gc_period_ms: u32,
+    pub shipper_throttle_packet_delay_ms: u32,
 }
 
 impl Default for Config {
@@ -61,6 +62,7 @@ impl Default for Config {
             watched_directory: None,
             disk_usage: 1024 * 1024,
             gc_period_ms: 10_000,
+            shipper_throttle_packet_delay_ms: 0,
         }
     }
 }

myceli/src/listener.rs

@@ -48,6 +48,7 @@ impl<T: Transport + Send + 'static> Listener<T> {
         shipper_timeout_duration: u64,
         shipper_window_size: u32,
         block_size: u32,
+        shipper_packet_delay_ms: u32,
     ) -> Result<()> {
         // First setup the shipper and its pieces
         let (shipper_sender, shipper_receiver) = mpsc::channel();
@@ -67,6 +68,7 @@ impl<T: Transport + Send + 'static> Listener<T> {
                 initial_connected,
                 block_size,
                 shipper_radio,
+                shipper_packet_delay_ms
             )
             .expect("Shipper creation failed");
             shipper.receive_msg_loop();

myceli/src/main.rs

@@ -52,7 +52,7 @@ fn main() -> Result<()> {
     .expect("Listener creation failed");
     println!("pid={}", std::process::id());
     listener
-        .start(cfg.retry_timeout_duration, cfg.window_size, cfg.block_size)
+        .start(cfg.retry_timeout_duration, cfg.window_size, cfg.block_size,cfg.shipper_throttle_packet_delay_ms)
         .expect("Error encountered in listener operation");
     Ok(())
 }

myceli/src/shipper.rs

@@ -53,6 +53,7 @@ pub struct Shipper<T> {
     connected: Arc<Mutex<bool>>,
     // Radio address
     radio_address: Option<String>,
+    packet_delay_ms: u32,
 }
 
 impl<T: Transport + Send + 'static> Shipper<T> {
@@ -67,6 +68,7 @@ impl<T: Transport + Send + 'static> Shipper<T> {
         connected: Arc<Mutex<bool>>,
         block_size: u32,
         radio_address: Option<String>,
+        packet_delay_ms: u32,
     ) -> Result<Shipper<T>> {
         let storage = Storage::new(storage_provider, block_size);
         Ok(Shipper {
@@ -79,6 +81,7 @@ impl<T: Transport + Send + 'static> Shipper<T> {
             transport,
             connected,
             radio_address,
+            packet_delay_ms,
         })
     }
 
@@ -367,6 +370,9 @@ impl<T: Transport + Send + 'static> Shipper<T> {
 
         info!("Transmitting {msg:?} to {_resolved_target_addr}");
         self.transport.send(msg, target_addr)?;
+        if self.packet_delay_ms > 0 {
+            std::thread::sleep(Duration::from_millis(self.packet_delay_ms.into()));
+        }
         Ok(())
     }
 
@@ -382,6 +388,7 @@ impl<T: Transport + Send + 'static> Shipper<T> {
                 block.cid.to_string()
             );
             self.transmit_msg(Message::data_block(transmission), target_addr)?;
+
         }
 
         Ok(())
@@ -517,6 +524,7 @@ mod tests {
                 Arc::new(Mutex::new(true)),
                 BLOCK_SIZE,
                 None,
+                0
             )
             .unwrap();
             TestShipper {

myceli/tests/utils/mod.rs

@@ -99,7 +99,7 @@ fn start_listener_thread(listen_addr: SocketAddr, db_path: ChildPath) {
     let mut listener =
         Listener::new(&listen_addr, db_path, transport, BLOCK_SIZE, None, 9).unwrap();
     listener
-        .start(10, 2, BLOCK_SIZE)
+        .start(10, 2, BLOCK_SIZE, 0)
         .expect("Error encountered in listener");
 }