Input channel created on new

src/clients/mod.rs

@@ -12,8 +12,10 @@ pub mod validator;
 
 
 // TODO: put that in config once it exists
-const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // assume seconds
-const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; // assume seconds;
+const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0;
+// assume seconds
+const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0;
+// assume seconds;
 const FETCH_MESSAGES_DELAY: f64 = 10.0; // assume seconds;
 
 // provider-poller sends polls service provider; receives messages
@@ -38,32 +40,38 @@ impl MixTrafficController {
 
 pub struct NymClient {
     // to be replaced by something else I guess
-    address: DestinationAddressBytes
+    address: DestinationAddressBytes,
+    pub input_tx: mpsc::UnboundedSender<Vec<u8>>,
+    // to be used by "send" function or socket, etc
+    input_rx: mpsc::UnboundedReceiver<Vec<u8>>,
 }
 
 type TripleFutureResult = (Result<(), Box<dyn std::error::Error>>, Result<(), Box<dyn std::error::Error>>, Result<(), Box<dyn std::error::Error>>);
 
 impl NymClient {
     pub fn new(address: DestinationAddressBytes) -> Self {
+        let (input_tx, input_rx) = mpsc::unbounded::<Vec<u8>>();
+
         NymClient {
-            address
+            address,
+            input_tx,
+            input_rx,
         }
     }
 
-    async fn start_loop_cover_traffic_stream(&self, mut tx: mpsc::Sender<Vec<u8>>) -> Result<(), Box<dyn std::error::Error>> {
+    async fn start_loop_cover_traffic_stream(mut tx: mpsc::Sender<Vec<u8>>) -> Result<(), Box<dyn std::error::Error>> {
         loop {
             let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY);
             let delay_duration = Duration::from_secs_f64(delay);
             println!("waiting for {:?}", delay_duration);
             tokio::time::delay_for(delay_duration).await;
             println!("waited {:?} - time to send cover message!", delay_duration);
-            let dummy_message = vec![1,2,3];
-            tx.send(dummy_message).await;
+            let dummy_message = vec![1, 2, 3];
+            tx.send(dummy_message).await?;
         }
-
     }
 
-    async fn control_out_queue(&self) -> Result<(), Box<dyn std::error::Error>> {
+    async fn control_out_queue(mut mix_tx: mpsc::Sender<Vec<u8>>, mut input_rx: mpsc::UnboundedReceiver<Vec<u8>>) -> Result<(), Box<dyn std::error::Error>> {
         loop {
             println!("here I will be sending real traffic (or loop cover if nothing is available)");
             let delay_duration = Duration::from_secs_f64(10.0);
@@ -72,7 +80,7 @@ impl NymClient {
     }
 
 
-    async fn start_provider_polling(&self) -> Result<(), Box<dyn std::error::Error>> {
+    async fn start_provider_polling() -> Result<(), Box<dyn std::error::Error>> {
         loop {
             println!("here I will be polling provider for messages");
             let delay_duration = Duration::from_secs_f64(10.0);
@@ -81,20 +89,21 @@ impl NymClient {
     }
 
 
-    async fn start_traffic(&self) -> TripleFutureResult {
+    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
+        println!("starting nym client");
+
         let mix_chan_buf_size = 64;
         let (mix_tx, mix_rx) = mpsc::channel(mix_chan_buf_size);
 
-        tokio::spawn(MixTrafficController::run(mix_rx));
-        futures::future::join3(self.start_loop_cover_traffic_stream(mix_tx), self.control_out_queue(), self.start_provider_polling()).await
-    }
-
-    pub fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
-        println!("starting nym client");
         let mut rt = Runtime::new()?;
 
+        rt.spawn(MixTrafficController::run(mix_rx));
+
         rt.block_on(async {
-            let future_results = self.start_traffic().await;
+            let future_results = futures::future::join3(
+                NymClient::start_loop_cover_traffic_stream(mix_tx.clone()),
+                NymClient::control_out_queue(mix_tx, self.input_rx),
+                NymClient::start_provider_polling()).await;
             assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok());
         });