morethantext/src/queue.rs

use crate::{client::Request, field::Field};
use std::{
    collections::HashMap,
    sync::{
        mpsc::{channel, Receiver, Sender},
        Arc, Mutex, RwLock,
    },
    thread::spawn,
};
use uuid::Uuid;

#[derive(Clone)]
pub enum MsgType {
    ClientMessage,
    ClientRequest,
    NewClientMessage,
    NoOp,
}

#[derive(Clone)]
pub struct Message {
    id: Uuid,
    class: MsgType,
    data: HashMap<String, Field>,
}

impl Message {
    pub fn new(msg_type: MsgType) -> Self {
        Self {
            id: Uuid::nil(),
            class: msg_type,
            data: HashMap::new(),
        }
    }

    fn reply(&self, data: MsgType) -> Message {
        Self {
            id: self.id.clone(),
            class: data,
            data: self.data.clone(),
        }
    }

    pub fn get_class(&self) -> &MsgType {
        &self.class
    }

    pub fn add_data<S, F>(&mut self, name: S, data: F)
    where
        S: Into<String>,
        F: Into<Field>,
    {
        self.data.insert(name.into(), data.into());
    }

    pub fn get_data(&self) -> &HashMap<String, Field> {
        &self.data
    }
}

impl From<Request> for Message {
    fn from(value: Request) -> Self {
        let msg = Message::new(MsgType::ClientRequest);
        msg.reply(MsgType::ClientRequest)
    }
}

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

    #[test]
    fn new_message() {
        let msg = Message::new(MsgType::NoOp);
        assert_eq!(msg.id, Uuid::nil());
        match msg.class {
            MsgType::NoOp => (),
            _ => unreachable!("new defaults to noop"),
        }
        assert!(msg.data.is_empty());
    }

    #[test]
    fn create_reply() {
        let id = Uuid::new_v4();
        let mut msg = Message::new(MsgType::NoOp);
        msg.id = id.clone();
        let data = MsgType::NewClientMessage;
        let result = msg.reply(data);
        assert_eq!(result.id, id);
        match result.class {
            MsgType::NewClientMessage => {}
            _ => unreachable!("should have been a registration request"),
        }
    }

    #[test]
    fn get_message_type() {
        let msg = Message::new(MsgType::NoOp);
        match msg.get_class() {
            MsgType::NoOp => {}
            _ => unreachable!("should have bneen noopn"),
        }
    }

    #[test]
    fn add_data() {
        let mut msg = Message::new(MsgType::NoOp);
        let one = "one";
        let two = "two".to_string();
        msg.add_data(one, one);
        msg.add_data(two.clone(), two.clone());
        let result = msg.get_data();
        assert_eq!(result.get(one).unwrap().to_string(), one);
        assert_eq!(result.get(&two).unwrap().to_string(), two);
    }
}

struct ServiceRegistry {
    store: Arc<Mutex<Vec<Sender<Message>>>>,
}

impl ServiceRegistry {
    fn new() -> Self {
        Self {
            store: Arc::new(Mutex::new(Vec::new())),
        }
    }

    fn add(&self, tx: Sender<Message>) {
        let mut store = self.store.lock().unwrap();
        store.push(tx);
    }

    fn send(&self, msg: Message) {
        let mut store = self.store.lock().unwrap();
        for sender in store.iter() {
            sender.send(msg.clone()).unwrap();
        }
    }
}

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

    #[test]
    fn create_registry() {
        let reg = ServiceRegistry::new();
        let (tx1, rx1) = channel();
        let (tx2, rx2) = channel();
        reg.add(tx1);
        reg.add(tx2);
        reg.send(Message::new(MsgType::NoOp));
        rx1.recv().unwrap();
        rx2.recv().unwrap();
    }
}

struct Queue {
    registry: Arc<RwLock<Vec<Sender<Message>>>>,
    rx: Receiver<Message>,
}

impl Queue {
    fn new(rx: Receiver<Message>, registry: Arc<RwLock<Vec<Sender<Message>>>>) -> Self {
        Self {
            registry: registry,
            rx: rx,
        }
    }

    fn start(registry: Arc<RwLock<Vec<Sender<Message>>>>) -> Sender<Message> {
        let (tx, rx) = channel();
        spawn(move || {
            let mut queue = Queue::new(rx, registry);
            queue.listen();
        });
        tx
    }

    fn listen(&mut self) {
        loop {
            let mut msg = self.rx.recv().unwrap();
            msg.id = Uuid::new_v4();
            let senders = self.registry.read().unwrap();
            for sender in senders.iter() {
                sender.send(msg.reply(MsgType::ClientMessage)).unwrap();
            }
        }
    }
}

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

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

    fn start_queue() -> (Sender<Message>, Receiver<Message>) {
        let reg: Arc<RwLock<Vec<Sender<Message>>>> = Arc::new(RwLock::new(Vec::new()));
        let (tx, rx) = channel::<Message>();
        let mut data = reg.write().unwrap();
        data.push(tx);
        drop(data);
        let queue_tx = Queue::start(Arc::clone(&reg));
        (queue_tx, rx)
    }

    #[test]
    fn get_new_client_message() {
        let (tx, rx) = start_queue();
        let initial = Message::new(MsgType::NoOp);
        let msg = initial.reply(MsgType::NewClientMessage);
        tx.send(msg).unwrap();
        let msg = rx.recv_timeout(TIMEOUT).unwrap();
        match msg.class {
            MsgType::ClientMessage => assert_ne!(msg.id, initial.id),
            _ => unreachable!("should have been a client message"),
        }
    }

    #[test]
    fn new_client_messages_are_unique() {
        let (tx, rx) = start_queue();
        let msg = Message::new(MsgType::NoOp);
        let mut ids: Vec<Uuid> = Vec::new();
        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
            );
            ids.push(result.id);
        }
    }
}