morethantext/src/message.rs

use std::{
    collections::HashMap,
    sync::{
        mpsc::{channel, Receiver, Sender},
        Arc, RwLock,
    },
    thread::spawn,
};
use uuid::Uuid;

#[cfg(test)]
mod support_test {
    use std::time::Duration;

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

#[derive(Clone, Debug)]
enum MTTError {
    DocumentAlreadyExists(String),
    DocumentNotFound(String),
    RouteNoListeners,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
enum Action {
    Create,
    Query,
    Reply,
}

impl From<MsgAction> for Action {
    fn from(value: MsgAction) -> Self {
        match value {
            MsgAction::Create(_) => Action::Create,
            MsgAction::Query(_) => Action::Query,
            MsgAction::Reply(_) => Action::Reply,
        }
    }
}

impl From<&MsgAction> for Action {
    fn from(value: &MsgAction) -> Self {
        let action = value.clone();
        Self::from(action)
    }
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
enum NameID {
    ID(Uuid),
    Name(String),
}

impl From<&str> for NameID {
    fn from(value: &str) -> Self {
        Self::Name(value.to_string())
    }
}

impl From<String> for NameID {
    fn from(value: String) -> Self {
        Self::Name(value)
    }
}

impl From<Uuid> for NameID {
    fn from(value: Uuid) -> Self {
        Self::ID(value)
    }
}

impl From<&NameID> for NameID {
    fn from(value: &NameID) -> Self {
        value.clone()
    }
}

#[derive(Clone)]
enum MsgAction {
    Create(DocDef),
    Query(Access),
    Reply(Response),
}

#[derive(Clone)]
struct Message {
    msg_id: Uuid,
    document_id: NameID,
    action: MsgAction,
}

impl Message {
    fn new<D>(doc_id: D, action: MsgAction) -> Self
    where
        D: Into<NameID>,
    {
        Self {
            msg_id: Uuid::new_v4(),
            document_id: doc_id.into(),
            action: action,
        }
    }

    fn get_message_id(&self) -> &Uuid {
        &self.msg_id
    }

    fn get_document_id(&self) -> &NameID {
        &self.document_id
    }

    fn get_action(&self) -> &MsgAction {
        &self.action
    }

    fn reply(&self, resp: Response) -> Self {
        Self {
            msg_id: self.msg_id.clone(),
            document_id: self.document_id.clone(),
            action: MsgAction::Reply(resp),
        }
    }
}

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

    #[test]
    fn can_the_document_be_a_stringi_reference() {
        let dts = ["one", "two"];
        for document in dts.into_iter() {
            let msg = Message::new(document, MsgAction::Create(DocDef::new()));
            match msg.get_document_id() {
                NameID::ID(_) => unreachable!("should have been a string id"),
                NameID::Name(data) => assert_eq!(data, document),
            }
            match msg.get_action() {
                MsgAction::Create(_) => {}
                _ => unreachable!("should have been a create document"),
            }
        }
    }

    #[test]
    fn can_the_document_be_a_string() {
        let dts = ["one".to_string(), "two".to_string()];
        for document in dts.into_iter() {
            let msg = Message::new(document.clone(), MsgAction::Query(Access::new()));
            match msg.get_document_id() {
                NameID::ID(_) => unreachable!("should have been a string id"),
                NameID::Name(data) => assert_eq!(data, &document),
            }
            match msg.get_action() {
                MsgAction::Query(_) => {}
                _ => unreachable!("should have been an access query"),
            }
        }
    }

    #[test]
    fn can_the_document_be_an_id() {
        let document = Uuid::new_v4();
        let msg = Message::new(document.clone(), MsgAction::Query(Access::new()));
        match msg.get_document_id() {
            NameID::ID(data) => assert_eq!(data, &document),
            NameID::Name(_) => unreachable!("should have been an id"),
        }
        match msg.get_action() {
            MsgAction::Query(_) => {}
            _ => unreachable!("should have been an access query"),
        }
    }

    #[test]
    fn is_the_message_id_random() {
        let mut ids: Vec<Uuid> = Vec::new();
        for _ in 0..5 {
            let msg = Message::new("tester", MsgAction::Create(DocDef::new()));
            let id = msg.get_message_id().clone();
            assert!(!ids.contains(&id), "{:?} containts  {}", ids, id);
            ids.push(id);
        }
    }

    #[test]
    fn Can_make_reply_message() {
        let name = "testing";
        let msg = Message::new(name, MsgAction::Query(Access::new()));
        let responce = Response::new();
        let reply = msg.reply(responce);
        assert_eq!(reply.get_message_id(), msg.get_message_id());
        match reply.get_document_id() {
            NameID::Name(data) => assert_eq!(data, name),
            _ => unreachable!("should have been a name"),
        }
        match reply.get_action() {
            MsgAction::Reply(_) => {}
            _ => unreachable!("should have been a reply"),
        }
    }
}

#[derive(Clone, Debug)]
enum Include<T> {
    All,
    Some(T),
}

impl<T: PartialEq> PartialEq for Include<T> {
    fn eq(&self, other: &Self) -> bool {
        match self {
            Include::All => true,
            Include::Some(data) => match other {
                Include::All => true,
                Include::Some(other_data) => data == other_data,
            },
        }
    }
}

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

    #[test]
    fn does_all_equal_evberything() {
        let a: Include<isize> = Include::All;
        let b: Include<isize> = Include::Some(5);
        let c: Include<isize> = Include::Some(7);
        assert!(a == a, "all should equal all");
        assert!(a == b, "all should equal some");
        assert!(b == a, "some should equal all");
        assert!(b == b, "same some should equal");
        assert!(b != c, "different somes do not equal");
    }
}

#[derive(Clone, Eq, Hash, PartialEq)]
struct RouteID {
    action: Option<Action>,
    doc_type: Option<Uuid>,
    msg_id: Option<Uuid>,
}

impl From<Route> for RouteID {
    fn from(value: Route) -> Self {
        Self {
            action: match value.action {
                Include::All => None,
                Include::Some(action) => Some(action.clone()),
            },
            doc_type: match value.doc_type {
                Include::All => None,
                Include::Some(doc) => Some(doc.clone()),
            },
            msg_id: match value.msg_id {
                Include::All => None,
                Include::Some(id) => Some(id.clone()),
            },
        }
    }
}

#[derive(Clone, Debug, PartialEq)]
struct Route {
    action: Include<Action>,
    doc_type: Include<Uuid>,
    msg_id: Include<Uuid>,
}

impl Route {
    fn new(msg_id: Include<Uuid>, doc: Include<Uuid>, action: Include<Action>) -> Self {
        Self {
            action: action,
            doc_type: doc,
            msg_id: msg_id,
        }
    }
}

impl From<RouteID> for Route {
    fn from(value: RouteID) -> Self {
        Self {
            action: match value.action {
                Some(data) => Include::Some(data.clone()),
                None => Include::All,
            },
            doc_type: match value.doc_type {
                Some(doc) => Include::Some(doc.clone()),
                None => Include::All,
            },
            msg_id: match value.msg_id {
                Some(msg) => Include::Some(msg.clone()),
                None => Include::All,
            },
        }
    }
}

impl From<&RouteID> for Route {
    fn from(value: &RouteID) -> Self {
        Self {
            action: match &value.action {
                Some(data) => Include::Some(data.clone()),
                None => Include::All,
            },
            doc_type: match &value.doc_type {
                Some(doc) => Include::Some(doc.clone()),
                None => Include::All,
            },
            msg_id: match &value.msg_id {
                Some(msg) => Include::Some(msg.clone()),
                None => Include::All,
            },
        }
    }
}

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

    #[test]
    fn can_a_route_set_action() {
        let actions = [Action::Query, Action::Reply];
        for action in actions.into_iter() {
            let route = Route::new(Include::All, Include::All, Include::Some(action.clone()));
            match route.msg_id {
                Include::All => {}
                Include::Some(_) => unreachable!("should have been all"),
            }
            match route.doc_type {
                Include::All => {}
                Include::Some(_) => unreachable!("should have been all"),
            }
            match route.action {
                Include::All => unreachable!("should be a specific value"),
                Include::Some(result) => assert_eq!(result, action),
            }
        }
    }

    #[test]
    fn can_route_set_document_by_name() {
        let doc_id = Uuid::new_v4();
        let route = Route::new(Include::All, Include::Some(doc_id.clone()), Include::All);
        match route.msg_id {
            Include::All => {}
            Include::Some(_) => unreachable!("should have been all"),
        }
        match route.doc_type {
            Include::All => unreachable!("should be a specific value"),
            Include::Some(result) => assert_eq!(result, doc_id),
        }
        match route.action {
            Include::All => {}
            Include::Some(_) => unreachable!("should have been all"),
        }
    }

    #[test]
    fn can_route_set_document_by_id() {
        let id = Uuid::new_v4();
        let route = Route::new(Include::All, Include::Some(id.clone()), Include::All);
        match route.msg_id {
            Include::All => {}
            Include::Some(_) => unreachable!("should have been all"),
        }
        match route.doc_type {
            Include::All => unreachable!("should be a specific value"),
            Include::Some(result) => assert_eq!(result, id),
        }
        match route.action {
            Include::All => {}
            Include::Some(_) => unreachable!("should have been all"),
        }
    }

    #[test]
    fn can_route_be_set_by_message_id() {
        let id = Uuid::new_v4();
        let route = Route::new(Include::Some(id.clone()), Include::All, Include::All);
        match route.msg_id {
            Include::All => unreachable!("should be a specific value"),
            Include::Some(result) => assert_eq!(result, id),
        }
        match route.doc_type {
            Include::All => {}
            Include::Some(_) => unreachable!("should have been all"),
        }
        match route.action {
            Include::All => {}
            Include::Some(_) => unreachable!("should have been all"),
        }
    }
}

#[derive(Clone)]
struct RouteRequest {
    msg_id: Include<Uuid>,
    doc_name: Include<String>,
    action: Include<Action>,
}

impl RouteRequest {
    fn new(msg_id: Include<Uuid>, doc_name: Include<String>, action: Include<Action>) -> Self {
        Self {
            msg_id: msg_id,
            doc_name: doc_name,
            action: action,
        }
    }
}

struct QueueData {
    senders: HashMap<Uuid, Sender<Message>>,
    names: HashMap<String, Uuid>,
    routes: HashMap<RouteID, Vec<Uuid>>,
}

impl QueueData {
    fn new() -> Self {
        Self {
            senders: HashMap::new(),
            names: HashMap::new(),
            routes: HashMap::new(),
        }
    }

    fn get_doc_id<N>(&self, nameid: N) -> Result<Uuid, MTTError>
    where
        N: Into<NameID>,
    {
        let sender_id = match nameid.into() {
            NameID::Name(name) => match self.names.get(&name) {
                Some(id) => id.clone(),
                None => return Err(MTTError::DocumentNotFound(name.clone())),
            },
            NameID::ID(id) => id.clone(),
        };
        if self.senders.contains_key(&sender_id) {
            Ok(sender_id)
        } else {
            Err(MTTError::DocumentNotFound(sender_id.to_string()))
        }
    }

    fn register(
        &mut self,
        tx: Sender<Message>,
        name: String,
        routes: Vec<RouteRequest>,
    ) -> Result<(), MTTError> {
        let mut id = Uuid::new_v4();
        while self.senders.contains_key(&id) {
            id = Uuid::new_v4();
        }
        match self.get_doc_id(name.clone()) {
            Ok(_) => return Err(MTTError::DocumentAlreadyExists(name)),
            Err(_) => {}
        }
        let mut holder: HashMap<RouteID, Vec<Uuid>> = HashMap::new();
        for route in routes.iter() {
            let doc_type = match &route.doc_name {
                Include::Some(doc_name) => {
                    if doc_name == &name {
                        Include::Some(id.clone())
                    } else {
                        match self.get_doc_id(doc_name.to_string()) {
                            Ok(doc_id) => Include::Some(doc_id),
                            Err(err) => return Err(err),
                        }
                    }
                }
                Include::All => Include::All,
            };
            let route_id: RouteID =
                Route::new(route.msg_id.clone(), doc_type, route.action.clone()).into();
            match self.routes.get(&route_id) {
                Some(senders) => {
                    let mut addition = senders.clone();
                    addition.push(id.clone());
                    holder.insert(route_id, addition);
                }
                None => {
                    let senders = [id.clone()].to_vec();
                    holder.insert(route_id, senders);
                }
            }
        }
        self.senders.insert(id.clone(), tx);
        self.names.insert(name.clone(), id.clone());
        for (route_id, senders) in holder.iter() {
            self.routes.insert(route_id.clone(), senders.clone());
        }
        Ok(())
    }

    fn send(&self, msg: Message) -> Result<(), MTTError> {
        let doc_id: Include<Uuid> = match self.get_doc_id(msg.get_document_id()) {
            Ok(id) => Include::Some(id.clone()),
            Err(err) => {
                let action: Action = msg.get_action().into();
                if action == Action::Create {
                    Include::Some(Uuid::nil())
                } else {
                    return Err(err);
                }
            }
        };
        let route = Route::new(
            Include::Some(msg.get_message_id().clone()),
            doc_id,
            Include::Some(msg.get_action().into()),
        );
        for (send_route, send_ids) in self.routes.iter() {
            if route == send_route.into() {
                for send_id in send_ids {
                    let tx = self.senders.get(&send_id).unwrap();
                    tx.send(msg.clone()).unwrap();
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod queuedatas {
    use super::support_test::TIMEOUT;
    use super::*;
    use std::sync::mpsc::RecvTimeoutError;

    #[test]
    fn can_document_be_registered() {
        let mut queuedata = QueueData::new();
        let (tx, rx) = channel();
        let name = Uuid::new_v4().to_string();
        let routes = [
            RouteRequest::new(
                Include::All,
                Include::Some(name.clone()),
                Include::Some(Action::Query),
            ),
            RouteRequest::new(
                Include::All,
                Include::Some(name.clone()),
                Include::Some(Action::Reply),
            ),
        ]
        .to_vec();
        queuedata.register(tx, name.clone(), routes).unwrap();
        let msg1 = Message::new(name.clone(), MsgAction::Query(Access::new()));
        let msg2 = Message::new(name.clone(), MsgAction::Reply(Response::new()));
        let msg3 = Message::new(name.clone(), MsgAction::Create(DocDef::new()));
        queuedata.send(msg1.clone()).unwrap();
        queuedata.send(msg2.clone()).unwrap();
        queuedata.send(msg3.clone()).unwrap();
        let result1 = rx.recv_timeout(TIMEOUT).unwrap();
        let result2 = rx.recv_timeout(TIMEOUT).unwrap();
        match rx.recv_timeout(TIMEOUT) {
            Ok(_) => unreachable!("should have timed out"),
            Err(err) => match err {
                RecvTimeoutError::Timeout => {}
                _ => unreachable!("should have timed out"),
            },
        }
        assert_eq!(result1.get_message_id(), msg1.get_message_id());
        assert_eq!(result2.get_message_id(), msg2.get_message_id());
        match result1.get_action() {
            MsgAction::Query(_) => {}
            _ => unreachable!("should have been a query"),
        }
        match result2.get_action() {
            MsgAction::Reply(_) => {}
            _ => unreachable!("should have been a query"),
        }
    }

    #[test]
    fn does_register_fail_on_duplicate_documents() {
        let mut queuedata = QueueData::new();
        let (tx1, _) = channel();
        let (tx2, _) = channel();
        let name = Uuid::new_v4().to_string();
        queuedata
            .register(tx1, name.to_string(), Vec::new())
            .unwrap();
        match queuedata.register(tx2, name.to_string(), Vec::new()) {
            Ok(_) => unreachable!("duplicates should create an error"),
            Err(err) => match err {
                MTTError::DocumentAlreadyExists(result) => assert_eq!(result, name),
                _ => unreachable!("should have been document does not exists"),
            },
        }
    }

    #[test]
    fn does_bad_route_prevent_registration() {
        let mut queuedata = QueueData::new();
        let (tx, rx) = channel();
        let good = "good";
        let bad = Uuid::new_v4().to_string();
        let routes = [
            RouteRequest::new(
                Include::All,
                Include::Some(good.to_string()),
                Include::Some(Action::Query),
            ),
            RouteRequest::new(
                Include::All,
                Include::Some(bad.clone()),
                Include::Some(Action::Reply),
            ),
        ]
        .to_vec();
        match queuedata.register(tx, good.to_string(), routes) {
            Ok(_) => unreachable!("should produce an error"),
            Err(err) => match err {
                MTTError::DocumentNotFound(result) => assert_eq!(result, bad),
                _ => unreachable!("Shouuld be document not found"),
            },
        }
        assert_eq!(queuedata.senders.len(), 0, "should not add to senders");
        assert_eq!(queuedata.names.len(), 0, "should not add to names");
        assert_eq!(queuedata.routes.len(), 0, "should nor add to routes");
    }

    #[test]
    fn is_sender_only_added_once_to_routes() {
        let mut queuedata = QueueData::new();
        let (tx, rx) = channel();
        let name = "project";
        let routes = [
            RouteRequest::new(
                Include::All,
                Include::Some(name.to_string()),
                Include::Some(Action::Query),
            ),
            RouteRequest::new(
                Include::All,
                Include::Some(name.to_string()),
                Include::Some(Action::Query),
            ),
        ]
        .to_vec();
        queuedata.register(tx, name.to_string(), routes).unwrap();
        for senders in queuedata.routes.values() {
            assert_eq!(senders.len(), 1, "should be no double entries");
        }
    }

    #[test]
    fn does_a_bad_document_name_fail() {
        let docname = Uuid::new_v4().to_string();
        let queuedata = QueueData::new();
        let msg = Message::new(docname.clone(), MsgAction::Query(Access::new()));
        match queuedata.send(msg) {
            Ok(_) => unreachable!("should have been an error"),
            Err(data) => match data {
                MTTError::DocumentNotFound(doc) => assert_eq!(doc, docname),
                _ => unreachable!("should have been a not found error"),
            },
        }
    }

    #[test]
    fn is_send_okay_if_no_one_is_listening() {
        let mut queuedata = QueueData::new();
        let name = "something";
        let (tx, _) = channel();
        queuedata
            .register(tx, name.to_string(), Vec::new())
            .unwrap();
        let msg = Message::new("something", MsgAction::Create(DocDef::new()));
        match queuedata.send(msg) {
            Ok(_) => {}
            Err(err) => unreachable!("got {:?}: should not error", err),
        }
    }

    #[test]
    fn can_more_than_one_document_respond() {
        let mut queuedata = QueueData::new();
        let name1 = "task";
        let name2 = "work";
        let action = MsgAction::Query(Access::new());
        let (tx1, rx1) = channel();
        let (tx2, rx2) = channel();
        let routes = [RouteRequest::new(
            Include::All,
            Include::Some(name1.to_string()),
            Include::All,
        )]
        .to_vec();
        queuedata
            .register(tx1, name1.to_string(), routes.clone())
            .unwrap();
        queuedata
            .register(tx2, name2.to_string(), routes.clone())
            .unwrap();
        let msg = Message::new(name1, action.clone());
        queuedata.send(msg.clone()).unwrap();
        let result1 = rx1.recv_timeout(TIMEOUT).unwrap();
        let result2 = rx2.recv_timeout(TIMEOUT).unwrap();
        assert_eq!(result1.get_message_id(), msg.get_message_id());
        assert_eq!(result1.get_message_id(), result2.get_message_id());
    }
}

#[derive(Clone)]
struct Queue {
    queue_data: Arc<RwLock<QueueData>>,
}

impl Queue {
    fn new() -> Self {
        Self {
            queue_data: Arc::new(RwLock::new(QueueData::new())),
        }
    }

    fn register(
        &mut self,
        tx: Sender<Message>,
        name: String,
        routes: Vec<RouteRequest>,
    ) -> Result<(), MTTError> {
        let mut queuedata = self.queue_data.write().unwrap();
        queuedata.register(tx, name, routes)
    }

    fn send(&self, msg: Message) -> Result<(), MTTError> {
        let queuedata = self.queue_data.read().unwrap();
        queuedata.send(msg)
    }
}

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

    #[test]
    fn create_a_queue() {
        Queue::new();
    }
}

struct CreateDoc {
    queue: Queue,
    rx: Receiver<Message>,
}

impl CreateDoc {
    fn new(queue: Queue, rx: Receiver<Message>) -> Self {
        Self {
            queue: queue,
            rx: rx,
        }
    }

    fn start(mut queue: Queue) {
        let (tx, rx) = channel();
        let routes = [RouteRequest::new(
            Include::All,
            Include::All,
            Include::Some(Action::Create),
        )]
        .to_vec();
        let id = queue.register(tx, "document".to_string(), routes).unwrap();
        let doc = CreateDoc::new(queue, rx);
        spawn(move || {
            doc.listen();
        });
    }

    fn listen(&self) {
        loop {
            let msg = self.rx.recv().unwrap();
            match msg.get_document_id() {
                NameID::Name(name) => Document::start(self.queue.clone(), name.clone()),
                NameID::ID(_) => unreachable!("should be a name"),
            }
            self.queue.send(msg.reply(Response::new())).unwrap();
        }
    }
}

#[derive(Clone)]
struct DocDef;

impl DocDef {
    fn new() -> Self {
        Self {}
    }
}

#[derive(Clone)]
struct Access;

impl Access {
    fn new() -> Self {
        Self {}
    }
}

#[derive(Clone)]
struct Response;

impl Response {
    fn new() -> Self {
        Self {}
    }
}

struct Document {
    queue: Queue,
    rx: Receiver<Message>,
}

impl Document {
    fn new(queue: Queue, rx: Receiver<Message>) -> Self {
        Self {
            queue: queue,
            rx: rx,
        }
    }

    fn start(mut queue: Queue, name: String) {
        let (tx, rx) = channel();
        queue.register(tx, name, Vec::new());
        let doc = Document::new(queue, rx);
        spawn(move || {
            doc.listen();
        });
    }

    fn listen(&self) {
        loop {
            self.rx.recv().unwrap();
        }
    }
}

#[cfg(test)]
mod createdocs {
    use super::support_test::TIMEOUT;
    use super::*;

    #[test]
    fn create_document_creation() {
        let mut queue = Queue::new();
        let (tx, rx) = channel();
        let routes = [RouteRequest::new(
            Include::All,
            Include::All,
            Include::Some(Action::Reply),
        )]
        .to_vec();
        queue.register(tx, "testing".to_string(), routes).unwrap();
        CreateDoc::start(queue.clone());
        let name = "project";
        let msg = Message::new(name, MsgAction::Create(DocDef::new()));
        queue.send(msg).unwrap();
        rx.recv_timeout(TIMEOUT).unwrap();
        let msg2 = Message::new(name, MsgAction::Query(Access::new()));
        queue.send(msg2).unwrap();
    }
}