From 58cfbef65ef73cc6b0c3c86580b8e1de06e2db4e Mon Sep 17 00:00:00 2001
From: Brandon Thomas <echelon@gmail.com>
Date: Sun, 13 Oct 2019 20:51:18 -0400
Subject: [PATCH] sending audio, but still very broken

---
 src/ipc.rs  | 78 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/main.rs | 75 +++++++++++++++++++++++++++++++++++++++++++++------
 2 files changed, 145 insertions(+), 8 deletions(-)

diff --git a/src/ipc.rs b/src/ipc.rs
index 4248f54832..452945899c 100644
--- a/src/ipc.rs
+++ b/src/ipc.rs
@@ -1,9 +1,47 @@
 use zmq;
 use zmq::{Error, Socket};
+use std::collections::VecDeque;
+use std::sync::{RwLock, PoisonError};
+
+pub struct AudioQueue {
+  queue: RwLock<VecDeque<i16>>,
+}
+
+impl AudioQueue {
+  pub fn new() -> Self {
+    Self {
+      queue: RwLock::new(VecDeque::new()),
+    }
+  }
+
+  pub fn push_back(&self, val: i16) {
+    match self.queue.write() {
+      Ok(mut queue) => {
+        queue.push_back(val);
+      },
+      Err(_) => {
+        unreachable!("This shouldn't happen");
+      },
+    }
+  }
+
+  pub fn drain(&self) -> Vec<i16> {
+    match self.queue.write() {
+      Ok(mut queue) => {
+        return queue.drain(..)
+            .collect::<Vec<_>>();
+      },
+      Err(_) => {
+        unreachable!("This shouldn't happen (drain)");
+      },
+    }
+  }
+}
 
 pub struct QueueSender {
   context: zmq::Context,
   socket: Option<zmq::Socket>,
+  queue: RwLock<VecDeque<i16>>,
 }
 
 impl QueueSender {
@@ -11,9 +49,33 @@ impl QueueSender {
     Self {
       context: zmq::Context::new(),
       socket: None,
+      queue: RwLock::new(VecDeque::new()),
     }
   }
 
+  /*pub fn push_back(&self, val: i16) {
+    match self.queue.write() {
+      Ok(mut queue) => {
+        queue.push_back(val);
+      },
+      Err(_) => {
+        unreachable!("This shouldn't happen (push_back)");
+      },
+    }
+  }*/
+
+  /*pub fn drain(&self) -> Vec<i16> {
+    match self.queue.write() {
+      Ok(mut queue) => {
+        return queue.drain(..)
+            .collect::<Vec<_>>();
+      },
+      Err(_) => {
+        unreachable!("This shouldn't happen (drain)");
+      },
+    }
+  }*/
+
   /// Connect to ZeroMQ server.
   pub fn connect(&mut self) -> bool {
     let mut socket = match self.context.socket(zmq::REQ) {
@@ -25,9 +87,11 @@ impl QueueSender {
 
     match socket.connect("tcp://127.0.0.1:5555") {
       Err(_) => {
+        println!("X");
         return false;
       },
       Ok(_) => {
+        println!("Y");
         self.socket = Some(socket);
         return true;
       },
@@ -40,12 +104,14 @@ impl QueueSender {
   {
     let mut socket = match &mut self.socket {
       None => {
+        println!("A");
         return false;
       },
       Some(s) => s,
     };
     match socket.send(data, 0) {
       Err(_) => {
+        println!("B");
         return false;
       },
       Ok(_) => {
@@ -53,4 +119,16 @@ impl QueueSender {
       },
     }
   }
+
+  /*
+    Python downsampled data,
+    Downsampled.shape: (31208,)
+    Downsampled.max: 0.2798332135682282
+    Downsampled.min: -0.26734864508137157
+    Downsampled.mean: 0.00024197357508030375
+  */
+  pub fn enqueue(&mut self, val: i16) -> bool {
+    let bytes = val.to_be_bytes();
+    self.send(&bytes[..])
+  }
 }
diff --git a/src/main.rs b/src/main.rs
index 9a7d56c57f..75a3483208 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -30,7 +30,8 @@ use tensorflow::Tensor;
 use tensorflow::version;
 use zmq::{Error, Socket};
 
-use ipc::QueueSender;
+use ipc::{QueueSender, AudioQueue};
+use std::sync::Arc;
 
 const INPUT_NAME : &'static str = "input_A_test";
 const OUTPUT_NAME : &'static str = "generator_A2B_3/output_transpose";
@@ -180,25 +181,83 @@ fn run_audio() -> Result<(), AudioError> {
 
     println!("Done");
 
-    let mut buffer = VecDeque::new();
+    //let mut buffer = VecDeque::new();
 
-    let mut queue_sender = QueueSender::new();
+    let mut audio_queue = Arc::new(AudioQueue::new());
+    let mut audio_queue_2 = audio_queue.clone();
+
+    thread::spawn(move || {
+      //let mut queue_sender = QueueSender::new();
+      //queue_sender.connect();
+
+      let mut context = zmq::Context::new();
+      let mut socket = context.socket(zmq::REQ).unwrap();
+
+      /*let mut socket = ctx.socket(zmq::REQ).unwrap();
+      socket.connect("tcp://127.0.0.1:1234").unwrap();
+      socket.send("hello world!", 0).unwrap();*/
+
+      socket.connect("tcp://127.0.0.1:5555").unwrap();
+
+      let mut reconnect = false;
+
+      loop {
+        let mut drained = audio_queue_2.drain();
+        println!("Len drained: {}", drained.len());
+
+        for val in drained {
+          let bytes = val.to_be_bytes();
+          loop {
+            if reconnect {
+              reconnect = false;
+
+              socket = match context.socket(zmq::REQ) {
+                Ok(s) => s,
+                Err(e) => {
+                  println!("Error creating socket: {:?}", e);
+                  continue
+                },
+              };
+
+              match socket.connect("tcp://127.0.0.1:5555") {
+                Ok(_) => {
+                  println!("B");
+                },
+                Err(err) => {
+                  println!("Err B: {:?}", err);
+                },
+              }
+
+            }
+            match socket.send(&bytes[..], 0) {
+              Ok(_) => { break; },
+              Err(e) => {
+                println!("send err: {:?}", e);
+                reconnect = true;
+              },
+            }
+
+          }
+        }
+      }
+    });
 
     loop {
       mic.record(&mut |_index, l, r| {
-        if !queue_sender.send("test hello") {
+        audio_queue.push_back(l);
+        /*if !queue_sender.enqueue(l) {
           queue_sender.connect();
-        }
-        buffer.push_back((l, l));
+        }*/
+        //buffer.push_back((l, l));
       });
 
-      speaker.play(&mut || {
+      /*speaker.play(&mut || {
         if let Some((lsample, rsample)) = buffer.pop_front() {
           AudioSample::stereo(lsample, rsample)
         } else {
           AudioSample::stereo(0, 0)
         }
-      });
+      });*/
     }
 }