morethantext/src/document.rs

use crate::{
    queue::{Message, MsgType, Queue},
    ActionType, ErrorType,
};
use serde_json::Value;
use std::{
    collections::HashMap,
    sync::mpsc::{channel, Receiver},
    thread::spawn,
};

const RESPONDS_TO: [MsgType; 1] = [MsgType::DocumentRequest];

pub struct Document {
    data: HashMap<String, String>,
    queue: Queue,
    rx: Receiver<Message>,
}

impl Document {
    fn new(queue: Queue, rx: Receiver<Message>) -> Self {
        let mut data = HashMap::new();
        data.insert("root".to_string(), "Something goes here.".to_string());
        Self {
            data: data,
            queue: queue,
            rx: rx,
        }
    }

    pub fn start(queue: Queue) {
        let (tx, rx) = channel();
        let mut document = Document::new(queue, rx);
        document.queue.add(tx, RESPONDS_TO.to_vec());
        spawn(move || {
            document.listen();
        });
    }

    fn listen(&mut self) {
        loop {
            let msg = self.rx.recv().unwrap();
            match msg.get_data("action") {
                Some(action) => match action.to_action().unwrap() {
                    ActionType::Add => self.add(msg),
                    _ => self.get(msg),
                },
                None => self.get(msg),
            }
        }
    }

    fn add(&mut self, msg: Message) {
        let name = msg.get_data("name").unwrap().to_string();
        match self.data.get(&name) {
            Some(_) => {
                let mut reply = msg.reply(MsgType::Error);
                reply.add_data("error_type", ErrorType::DocumentAlreadyExists);
                self.queue.send(reply).unwrap();
                return;
            }
            None => {}
        }
        let doc: Value = match serde_json::from_str(&msg.get_data("doc").unwrap().to_string()) {
            Ok(value) => value,
            Err(_) => {
                let mut reply = msg.reply(MsgType::Error);
                reply.add_data("error_type", ErrorType::DocumentInvalidRequest);
                self.queue.send(reply).unwrap();
                return;
            }
        };
        self.data
            .insert(name, doc["template"].as_str().unwrap().to_string());
        self.queue.send(msg.reply(MsgType::ActionOk)).unwrap();
    }

    fn get(&self, msg: Message) {
        let name = match msg.get_data("name") {
            Some(doc) => doc.to_string(),
            None => "root".to_string(),
        };
        let reply = match self.data.get(&name) {
            Some(data) => {
                let mut holder = msg.reply(MsgType::Document);
                holder.add_data("doc", data.clone());
                holder
            }
            None => {
                let mut holder = msg.reply(MsgType::Error);
                holder.add_data("error_type", ErrorType::DocumentNotFound);
                holder
            }
        };
        self.queue.send(reply).unwrap();
    }
}

#[cfg(test)]
pub mod documents {
    use super::*;
    use serde_json::json;
    use std::time::Duration;
    use uuid::Uuid;

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

    fn setup_document() -> (Queue, Receiver<Message>) {
        let queue = Queue::new();
        let (tx, rx) = channel();
        queue.add(
            tx,
            [MsgType::ActionOk, MsgType::Document, MsgType::Error].to_vec(),
        );
        Document::start(queue.clone());
        (queue, rx)
    }

    #[test]
    fn start_service() {
        let (queue, rx) = setup_document();
        let id = Uuid::new_v4();
        let mut msg = Message::new(MsgType::DocumentRequest);
        msg.add_data("sess_id", id.clone());
        queue.send(msg.clone()).unwrap();
        let reply = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply.get_id(), msg.get_id());
        match reply.get_msg_type() {
            MsgType::Document => {}
            _ => unreachable!("got {:?} should have gotten document", msg.get_msg_type()),
        }
        assert!(reply.get_data("doc").is_some());
    }

    #[test]
    fn no_existing_document() {
        let (queue, rx) = setup_document();
        let name = format!("name-{}", Uuid::new_v4());
        let mut msg = Message::new(MsgType::DocumentRequest);
        msg.add_data("name", name.clone());
        queue.send(msg.clone()).unwrap();
        let reply = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply.get_id(), msg.get_id());
        match reply.get_msg_type() {
            MsgType::Error => {}
            _ => unreachable!("got {:?}: shoud have been error", reply.get_msg_type()),
        }
        match reply.get_data("error_type") {
            Some(err) => match err.to_error_type().unwrap() {
                ErrorType::DocumentNotFound => {}
                _ => unreachable!("got {:?}: should have been document not found'", err),
            },
            None => unreachable!("should contain error type"),
        }
    }

    #[test]
    fn root_always_exists() {
        let (queue, rx) = setup_document();
        let mut msg = Message::new(MsgType::DocumentRequest);
        msg.add_data("name", "root");
        queue.send(msg).unwrap();
        let reply = rx.recv_timeout(TIMEOUT).unwrap();
        match reply.get_msg_type() {
            MsgType::Document => {}
            _ => unreachable!(
                "Got '{:?}': should have been a document",
                reply.get_msg_type()
            ),
        }
    }

    #[test]
    fn add_new_document() {
        let (queue, rx) = setup_document();
        let name = format!("name-{}", Uuid::new_v4());
        let content = format!("content-{}", Uuid::new_v4());
        let input = json!({
            "template": content.clone()
        });
        let mut msg1 = Message::new(MsgType::DocumentRequest);
        msg1.add_data("name", name.clone());
        msg1.add_data("action", ActionType::Add);
        msg1.add_data("doc", input.to_string());
        queue.send(msg1.clone()).unwrap();
        let reply1 = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply1.get_id(), msg1.get_id());
        match reply1.get_msg_type() {
            MsgType::ActionOk => {}
            _ => unreachable!(
                "got {:?}: should have received action ok",
                reply1.get_msg_type()
            ),
        }
        let mut msg2 = Message::new(MsgType::DocumentRequest);
        msg2.add_data("name", name.clone());
        msg2.add_data("action", ActionType::Get);
        queue.send(msg2.clone()).unwrap();
        let reply2 = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply2.get_id(), msg2.get_id());
        match reply2.get_msg_type() {
            MsgType::Document => {}
            _ => unreachable!(
                "got {:?}: should have received document",
                reply2.get_msg_type()
            ),
        }
        assert_eq!(reply2.get_data("doc").unwrap().to_string(), content);
    }

    #[test]
    fn add_does_not_overwrite_existing() {
        let (queue, rx) = setup_document();
        let mut holder = Message::new(MsgType::DocumentRequest);
        holder.add_data("name", "root");
        holder.add_data("action", ActionType::Get);
        queue.send(holder.clone()).unwrap();
        let binding = rx.recv_timeout(TIMEOUT).unwrap();
        let expected = binding.get_data("doc").unwrap();
        let input = json!({
            "template": format!("content-{}", Uuid::new_v4())
        });
        let mut msg = Message::new(MsgType::DocumentRequest);
        msg.add_data("name", "root");
        msg.add_data("action", ActionType::Add);
        msg.add_data("doc", input.to_string());
        queue.send(msg.clone()).unwrap();
        let reply = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply.get_id(), msg.get_id());
        match reply.get_msg_type() {
            MsgType::Error => {}
            _ => unreachable!(
                "got '{:?}': should have received document",
                reply.get_msg_type()
            ),
        }
        match reply.get_data("error_type") {
            Some(err) => match err.to_error_type().unwrap() {
                ErrorType::DocumentAlreadyExists => {}
                _ => unreachable!("got {:?}: should have been document not found'", err),
            },
            None => unreachable!("should contain error type"),
        }
        queue.send(holder).unwrap();
        let binding = rx.recv_timeout(TIMEOUT).unwrap();
        let result = binding.get_data("doc").unwrap();
        assert_eq!(result.to_string(), expected.to_string());
    }

    #[test]
    fn invalid_json() {
        let (queue, rx) = setup_document();
        let mut msg = Message::new(MsgType::DocumentRequest);
        msg.add_data("action", ActionType::Add);
        msg.add_data("name", "doc");
        msg.add_data("doc", "Invalid json request.");
        queue.send(msg.clone()).unwrap();
        let reply = rx.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(reply.get_id(), msg.get_id());
        match reply.get_msg_type() {
            MsgType::Error => {}
            _ => unreachable!(
                "got '{:?}': should have received document",
                reply.get_msg_type()
            ),
        }
        match reply.get_data("error_type") {
            Some(err) => match err.to_error_type().unwrap() {
                ErrorType::DocumentInvalidRequest => {}
                _ => unreachable!("got {:?}: should have been bad request'", err),
            },
            None => unreachable!("should contain error type"),
        }
    }
}