Moved client responses to mutex.

src/client.rs

@@ -1,40 +1,28 @@
+use crate::{queue::Message, utils::GenID};
 use std::{
-    sync::mpsc::{channel, Receiver, Sender},
+    collections::HashMap,
+    sync::{
+        mpsc::{channel, Receiver, Sender},
+        Arc, Mutex,
+    },
     thread::spawn,
 };
+use uuid::Uuid;
 
-pub struct Request {
-    tx: Sender<Reply>,
-}
+pub struct Request;
 
 impl Request {
-    pub fn new(tx: Sender<Reply>) -> Self {
-        Self { tx: tx }
-    }
-
-    fn get_sender(&self) -> &Sender<Reply> {
-        &self.tx
+    pub fn new() -> Self {
+        Self {}
     }
 }
 
 #[cfg(test)]
 mod requests {
     use super::*;
-    use replies::create_reply;
 
-    pub fn create_request() -> (Request, Receiver<Reply>) {
-        let (tx, rx) = channel();
-        let req = Request::new(tx);
-        (req, rx)
-    }
-
-    #[test]
-    fn new_request() {
-        let (tx, rx) = channel();
-        let req = Request::new(tx);
-        let sender = req.get_sender();
-        sender.send(create_reply()).unwrap();
-        rx.recv().unwrap();
+    pub fn create_request() -> Request {
+        Request::new()
     }
 }
 
@@ -59,28 +47,145 @@ mod replies {
     }
 }
 
+#[derive(Clone)]
+pub struct ClientRegistry {
+    registry: Arc<Mutex<HashMap<Uuid, Sender<Reply>>>>,
+}
+
+impl ClientRegistry {
+    pub fn new() -> Self {
+        Self {
+            registry: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    fn create_storage() -> Arc<Mutex<HashMap<Uuid, Sender<Reply>>>> {
+        Arc::new(Mutex::new(HashMap::new()))
+    }
+
+    fn get_id<'a>(
+        gen: &mut impl Iterator<Item = Uuid>,
+        data: &HashMap<Uuid, Sender<Reply>>,
+    ) -> Uuid {
+        let mut id = gen.next().unwrap();
+        while data.contains_key(&id) {
+            id = gen.next().unwrap();
+        }
+        id.clone()
+    }
+
+    pub fn add(&mut self, tx: Sender<Reply>) -> Uuid {
+        let id = Uuid::new_v4();
+        let mut reg = self.registry.lock().unwrap();
+        let mut gen_id = GenID::new();
+        let id = ClientRegistry::get_id(&mut gen_id, &reg);
+        reg.insert(id.clone(), tx);
+        id
+    }
+
+    fn send(&mut self, id: &Uuid, msg: Reply) {
+        let mut reg = self.registry.lock().unwrap();
+        let tx = reg.get(id).unwrap();
+        tx.send(msg).unwrap();
+        reg.remove(id).unwrap();
+    }
+}
+
+#[cfg(test)]
+mod clientregistries {
+    use super::*;
+    use std::{
+        sync::mpsc::{channel, Receiver},
+        time::Duration,
+    };
+
+    static TIMEOUT: Duration = Duration::from_millis(500);
+
+    #[test]
+    fn create_client_registry() {
+        let reg = ClientRegistry::new();
+        let data = reg.registry.lock().unwrap();
+        assert!(data.is_empty(), "needs to create an empty hashmap");
+    }
+
+    #[test]
+    fn send_from_client() {
+        let mut reg = ClientRegistry::new();
+        let count = 10;
+        let mut rxs: HashMap<Uuid, Receiver<Reply>> = HashMap::new();
+        for _ in 0..count {
+            let (tx, rx) = channel::<Reply>();
+            let id = reg.add(tx);
+            rxs.insert(id, rx);
+        }
+        assert_eq!(rxs.len(), count, "should have been {} receivers", count);
+        for (id, rx) in rxs.iter() {
+            let msg = Reply {};
+            reg.send(id, msg);
+            rx.recv_timeout(TIMEOUT).unwrap();
+        }
+        let data = reg.registry.lock().unwrap();
+        assert!(data.is_empty(), "should remove sender after sending");
+    }
+
+    #[test]
+    fn prevent_duplicates() {
+        let mut reg = ClientRegistry::new();
+        let (tx, rx) = channel::<Reply>();
+        let existing = reg.add(tx);
+        let expected = Uuid::new_v4();
+        let ids = [existing.clone(), expected.clone()];
+        let data = reg.registry.lock().unwrap();
+        let result = ClientRegistry::get_id(&mut ids.into_iter(), &data);
+        assert_eq!(result, expected);
+    }
+}
+
+#[derive(Clone)]
+pub struct ClientLink;
+
+impl ClientLink {
+    fn new() -> Self {
+        Self {}
+    }
+
+    pub fn forward(&self, req: Request) -> Reply {
+        Reply {}
+    }
+}
+
+#[cfg(test)]
+mod clientlinks {
+    use super::*;
+
+    #[test]
+    fn create_client_link() {
+        ClientLink::new();
+    }
+}
+
 pub struct Client {
-    rx: Receiver<Request>,
+    rx: Receiver<Message>,
 }
 
 impl Client {
-    fn new(rx: Receiver<Request>) -> Self {
+    fn new(rx: Receiver<Message>) -> Self {
         Self { rx: rx }
     }
 
-    pub fn start() -> Sender<Request> {
+    pub fn start() -> ClientLink {
         let (tx, rx) = channel();
         spawn(move || {
             let client = Client::new(rx);
             client.listen();
         });
-        tx
+        ClientLink::new()
     }
 
     fn listen(&self) {
         loop {
             let req = self.rx.recv().unwrap();
-            req.get_sender().send(Reply {}).unwrap();
+            //req.get_sender().send(Reply {}).unwrap();
         }
     }
 }
@@ -92,9 +197,8 @@ mod clients {
 
     #[test]
     fn start_client() {
-        let tx = Client::start();
-        let (req, rx) = create_request();
-        tx.send(req).unwrap();
-        rx.recv().unwrap();
+        let link = Client::start();
+        let req = create_request();
+        link.forward(req);
     }
 }

src/lib.rs

@@ -1,8 +1,9 @@
 mod client;
 mod field;
 mod queue;
+mod utils;
 
-use client::{Client, Reply, Request};
+use client::{Client, ClientLink, Reply, Request};
 use field::Field;
 use queue::Message;
 use std::sync::{
@@ -12,26 +13,28 @@ use std::sync::{
 
 #[derive(Clone)]
 pub struct MoreThanText {
+    client_link: ClientLink,
     registry: Arc<RwLock<Vec<Sender<Message>>>>,
-    tx: Sender<Request>,
 }
 
 impl MoreThanText {
     pub fn new() -> Self {
-        let tx = Client::start();
         Self {
+            client_link: Client::start(),
             registry: Arc::new(RwLock::new(Vec::new())),
-            tx: tx,
         }
     }
 
-    pub fn request<F>(&self, _session: Option<F>) -> Reply
+    pub fn request<F>(&mut self, _session: Option<F>) -> Reply
     where
         F: Into<Field>,
     {
-        let (tx, rx) = channel();
+        let req = Request::new();
+        self.client_link.forward(req)
+        /*
         let req = Request::new(tx);
-        self.tx.send(req).unwrap();
+        self.tx.send(req.into()).unwrap();
         rx.recv().unwrap()
+        */
     }
 }

src/main.rs

@@ -40,7 +40,7 @@ async fn mtt_conn(jar: CookieJar, state: State<MoreThanText>) -> impl IntoRespon
     };
     let (tx, mut rx) = channel(5);
     spawn(async move {
-        tx.send(state.request(sid)).await.unwrap();
+        tx.send(state.clone().request(sid)).await.unwrap();
     });
     let reply = rx.recv().await.unwrap();
     let cookie = Cookie::build((SESSION_KEY, reply.get_session()));

src/queue.rs

@@ -1,4 +1,4 @@
-use crate::field::Field;
+use crate::{client::Request, field::Field};
 use std::{
     collections::HashMap,
     sync::{
@@ -11,6 +11,7 @@ use uuid::Uuid;
 
 enum MsgType {
     ClientMessage,
+    ClientRequest,
     NewClientMessage,
     NoOp,
 }
@@ -55,6 +56,13 @@ impl Message {
     }
 }
 
+impl From<Request> for Message {
+    fn from(value: Request) -> Self {
+        let msg = Message::new();
+        msg.reply(MsgType::ClientRequest)
+    }
+}
+
 #[cfg(test)]
 mod messages {
     use super::*;
@@ -178,7 +186,11 @@ mod queues {
         for _ in 0..10 {
             tx.send(msg.reply(MsgType::NewClientMessage)).unwrap();
             let result = rx.recv().unwrap();
-            assert!(!ids.contains(&result.id.clone()), "{} is a duplicate", &result.id);
+            assert!(
+                !ids.contains(&result.id.clone()),
+                "{} is a duplicate",
+                &result.id
+            );
             ids.push(result.id);
         }
     }

src/utils.rs

@@ -0,0 +1,17 @@
+use uuid::Uuid;
+
+pub struct GenID;
+
+impl GenID {
+    pub fn new() -> Self {
+        GenID {}
+    }
+}
+
+impl Iterator for GenID {
+    type Item = Uuid;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        Some(Uuid::new_v4())
+    }
+}