morethantext/src/client.rs

use crate::{
    field::Field,
    queue::{Message, MsgType, Queue},
    utils::GenID,
};
use std::{
    collections::HashMap,
    sync::{
        mpsc::{channel, Receiver, Sender},
        Arc, Mutex,
    },
    thread::spawn,
};
use uuid::Uuid;

const RESPONS_TO: [MsgType; 2] = [MsgType::Document, MsgType::SessionValidated];

pub struct Request {
    pub session: Option<Field>,
}

impl Request {
    pub fn new(session: Option<Field>) -> Self {
        Self { session: session }
    }
}

#[cfg(test)]
pub mod requests {
    use super::*;

    pub fn get_root_document() -> Request {
        Request::new(None)
    }

    pub fn get_root_with_session(sess_id: &Uuid) -> Request {
        Request::new(Some(sess_id.clone().into()))
    }

    pub fn get_root_document_eith_session<F>(id: F) -> Request
    where
        F: Into<Field>,
    {
        Request::new(Some(id.into()))
    }

    #[test]
    fn new_request_no_session() {
        let sess: Option<Field> = None;
        let req = Request::new(sess);
        assert!(req.session.is_none(), "should not have a session")
    }

    #[test]
    fn new_request_with_session() {
        let id = Uuid::new_v4();
        let req = Request::new(Some(id.into()));
        match req.session {
            Some(result) => assert_eq!(result.to_uuid().unwrap(), id),
            None => unreachable!("should contain a session"),
        }
    }
}

pub struct Reply {
    sess_id: Uuid,
    content: String,
}

impl Reply {
    fn new(sess_id: Uuid, content: String) -> Self {
        Self {
            sess_id: sess_id,
            content: content,
        }
    }

    pub fn get_session(&self) -> Uuid {
        self.sess_id.clone()
    }

    pub fn get_content(&self) -> String {
        self.content.clone()
    }
}

#[cfg(test)]
mod replies {
    use super::*;

    pub fn create_reply() -> Reply {
        Reply {
            sess_id: Uuid::new_v4(),
            content: "some text".to_string(),
        }
    }

    #[test]
    fn create_new_reply() {
        let sess_id = Uuid::new_v4();
        let txt = Uuid::new_v4().to_string();
        let reply = Reply::new(sess_id, txt.clone());
        assert_eq!(reply.get_session(), sess_id);
        assert_eq!(reply.get_content(), txt);
    }
}

#[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 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 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.remove(id).unwrap();
        tx.send(msg).unwrap();
    }
}

#[cfg(test)]
mod clientregistries {
    use super::*;
    use crate::client::replies::create_reply;
    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 = create_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 {
    tx: Sender<Message>,
    registry: ClientRegistry,
}

impl ClientLink {
    fn new(tx: Sender<Message>, registry: ClientRegistry) -> Self {
        Self {
            tx: tx,
            registry: registry,
        }
    }

    pub fn send(&mut self, req: Request) -> Receiver<Reply> {
        let (tx, rx) = channel();
        let mut msg: Message = req.into();
        let id = self.registry.add(tx);
        msg.add_data("tx_id", id);
        self.tx.send(msg).unwrap();
        rx
    }
}

#[cfg(test)]
mod clientlinks {
    use super::*;
    use crate::client::replies::create_reply;
    use std::time::Duration;

    static TIMEOUT: Duration = Duration::from_millis(500);

    #[test]
    fn create_client_link() {
        let (tx, rx) = channel();
        let mut registry = ClientRegistry::new();
        let mut link = ClientLink::new(tx, registry.clone());
        let req = Request::new(None);
        let rx_client = link.send(req);
        let msg = rx.recv_timeout(TIMEOUT).unwrap();
        match msg.get_msg_type() {
            MsgType::ClientRequest => {}
            _ => unreachable!("should have been a client request"),
        }
        match msg.get_data("tx_id") {
            Some(result) => {
                let id = result.to_uuid().unwrap();
                registry.send(&id, create_reply());
                rx_client.recv().unwrap();
            }
            None => unreachable!("should have had a seender id"),
        }
    }
}

pub struct Client {
    queue: Queue,
    registry: ClientRegistry,
    return_to: HashMap<Uuid, Message>,
    rx: Receiver<Message>,
}

impl Client {
    fn new(rx: Receiver<Message>, queue: Queue) -> Self {
        Self {
            queue: queue,
            registry: ClientRegistry::new(),
            return_to: HashMap::new(),
            rx: rx,
        }
    }

    pub fn start(queue: Queue) -> ClientLink {
        let (tx, rx) = channel();
        queue.add(tx.clone(), RESPONS_TO.to_vec());
        let mut client = Client::new(rx, queue);
        let link = ClientLink::new(tx, client.get_registry());
        spawn(move || {
            client.listen();
        });
        link
    }

    fn listen(&mut self) {
        loop {
            let msg = self.rx.recv().unwrap();
            match msg.get_msg_type() {
                MsgType::ClientRequest => self.client_request(msg),
                MsgType::Document => self.document(msg),
                MsgType::SessionValidated => self.session(msg),
                _ => unreachable!("Received message it did not understand"),
            }
        }
    }

    fn get_registry(&self) -> ClientRegistry {
        self.registry.clone()
    }

    fn client_request(&mut self, msg: Message) {
        self.return_to.insert(msg.get_id(), msg.clone());
        let mut reply = msg.reply(MsgType::SessionValidate);
        match msg.get_data("sess_id") {
            Some(sess_id) => reply.add_data("sess_id", sess_id.clone()),
            None => {}
        }
        self.queue.send(reply).unwrap();
    }

    fn session(&mut self, msg: Message) {
        let initial_msg = self.return_to.get_mut(&msg.get_id()).unwrap();
        let mut reply = msg.reply(MsgType::DocumentRequest);
        match msg.get_data("sess_id") {
            Some(sess_id) => {
                initial_msg.add_data("sess_id", sess_id.clone());
                reply.add_data("sess_id", sess_id.clone());
            }
            None => unreachable!("validated should always have an id"),
        }
        self.queue.send(reply).unwrap();
    }

    fn document(&mut self, msg: Message) {
        let initial_msg = self.return_to.remove(&msg.get_id()).unwrap();
        let tx_id = initial_msg.get_data("tx_id").unwrap().to_uuid().unwrap();
        let reply = Reply::new(
            initial_msg.get_data("sess_id").unwrap().to_uuid().unwrap(),
            msg.get_data("doc").unwrap().to_string(),
        );
        self.registry.send(&tx_id, reply);
    }
}

#[cfg(test)]
mod clients {
    use super::*;
    use requests::get_root_with_session;
    use std::time::Duration;

    static TIMEOUT: Duration = Duration::from_millis(500);

    #[test]
    fn start_client() {
        let sess_id1 = Uuid::new_v4();
        let sess_id2 = Uuid::new_v4();
        let doc = Uuid::new_v4().to_string();
        let (tx, rx) = channel();
        let queue = Queue::new();
        queue.add(
            tx,
            [MsgType::SessionValidate, MsgType::DocumentRequest].to_vec(),
        );
        let mut link = Client::start(queue.clone());
        let req = get_root_with_session(&sess_id1);
        let reply_rx = link.send(req);
        let send1 = rx.recv_timeout(TIMEOUT).unwrap();
        match send1.get_msg_type() {
            MsgType::SessionValidate => {}
            _ => unreachable!("should request session validation"),
        }
        assert_eq!(
            send1.get_data("sess_id").unwrap().to_uuid().unwrap(),
            sess_id1
        );
        assert!(send1.get_data("tx_id").is_none());
        let mut response = send1.reply_with_data(MsgType::SessionValidated);
        response.add_data("sess_id", sess_id2);
        queue.send(response).unwrap();
        let send2 = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(send2.get_id(), send1.get_id());
        match send2.get_msg_type() {
            MsgType::DocumentRequest => {}
            _ => unreachable!("should request session validation"),
        }
        assert_eq!(
            send2.get_data("sess_id").unwrap().to_uuid().unwrap(),
            sess_id2
        );
        let mut document = send2.reply(MsgType::Document);
        document.add_data("doc", doc.clone());
        queue.send(document).unwrap();
        let reply = reply_rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply.get_session(), sess_id2);
        assert_eq!(reply.get_content(), doc);
    }
}