diff --git a/src/data_director.rs b/src/data_director.rs index b79a3d6..dadc6df 100644 --- a/src/data_director.rs +++ b/src/data_director.rs @@ -473,11 +473,14 @@ impl DocRegistry { Err(err) => reg.response(RegMsg::Error(err)), }, RegMsg::AddRoute(path) => { - // let route = self.doc_names.path_to_route(path).unwrap(); - let route = self.path_to_route(path).unwrap(); - reg.response(RegMsg::RouteID( - self.routes.add(route, reg.get_sender_id().clone()), - )) + let response = match self.path_to_route(path) { + Ok(route) => { + let id = self.routes.add(route, reg.get_sender_id().clone()); + RegMsg::RouteID(id) + } + Err(err) => RegMsg::Error(err), + }; + reg.response(response) } RegMsg::GetNameID(name) => match self.doc_names.get_id(name) { Ok(id) => reg.response(RegMsg::DocumentNameID(id.clone())), diff --git a/src/document.rs b/src/document.rs index 3f50da8..253157f 100644 --- a/src/document.rs +++ b/src/document.rs @@ -2,3 +2,4 @@ pub mod clock; pub mod create; pub mod definition; pub mod field; +pub mod session; diff --git a/src/document/definition.rs b/src/document/definition.rs index 2a61c55..72d34ca 100644 --- a/src/document/definition.rs +++ b/src/document/definition.rs @@ -216,7 +216,7 @@ impl DocDef { Self::with_names(names) } - fn with_names(names: Vec) -> Self { + pub fn with_names(names: Vec) -> Self { let routes = vec![ PathAction::new( Path::new( diff --git a/src/document/session.rs b/src/document/session.rs new file mode 100644 index 0000000..d4b2f95 --- /dev/null +++ b/src/document/session.rs @@ -0,0 +1,345 @@ +use crate::{ + data_director::{Include, Path, RegMsg, Register}, + document::{ + create::IndexType, + definition::{DocDef, DocFuncType}, + field::FieldType, + }, + message::{Action, CalcValue, Calculation, Delete, Message, Operand, Query, Update}, + name::{Name, NameType}, + router::Queue, +}; +use std::{sync::mpsc::channel, time::Duration}; + +pub struct Session; + +impl Session { + fn doc_names() -> Vec { + let mut names = Vec::new(); + names.push(Name::english("session")); + names + } + + pub fn start(mut queue: Queue) { + let mut docdef = DocDef::with_names(Self::doc_names()); + + let mut calc = Calculation::new(Operand::Add); + calc.add_value(FieldType::DateTime).unwrap(); + calc.add_value(Duration::from_hours(1)).unwrap(); + + let name_id = Name::english("id"); + docdef.add_field(name_id.clone(), FieldType::Uuid); + docdef.set_default(&name_id, FieldType::Uuid).unwrap(); + docdef.add_index(&name_id, IndexType::Unique).unwrap(); + + let name_expire = Name::english("expire"); + docdef.add_field(name_expire.clone(), FieldType::DateTime); + docdef.set_default(&name_expire, calc.clone()).unwrap(); + + let mut update = Update::new(Query::new()); + update + .get_values_mut() + .add_field(name_expire.clone(), calc.clone()); + let path = Path::new( + Include::All, + Include::Just(Session::doc_names()[0].clone().into()), + Include::Just(Action::OnQuery), + ); + let query_action = DocFuncType::ExistingQuery(update.into()); + docdef.add_route(path, query_action); + + let mut delete_qry = Query::new(); + let mut delete_calc = Calculation::new(Operand::LessThan); + delete_calc.add_value(FieldType::DateTime).unwrap(); + delete_calc + .add_value(CalcValue::Existing(FieldType::DateTime)) + .unwrap(); + delete_qry.add(name_expire.clone(), delete_calc); + let delete = Delete::new(delete_qry); + let clock_path = Path::new( + Include::All, + Include::Just(Name::english("clock").into()), + Include::Just(Action::OnUpdate), + ); + let delete_func = DocFuncType::Trigger(delete.into()); + docdef.add_route(clock_path, delete_func); + + let (tx, rx) = channel(); + let sender_id = queue.add_sender(tx); + let msg = Message::new(NameType::None, docdef); + let msg_id = msg.get_message_id().clone(); + let path = Path::new( + Include::Just(msg_id), + Include::All, + Include::Just(Action::Reply), + ); + let reg_msg = Register::new(sender_id.clone(), RegMsg::AddRoute(path)); + queue.send(msg.response(reg_msg)).unwrap(); + rx.recv().unwrap(); + queue.send(msg).unwrap(); + rx.recv().unwrap(); + queue.remove_sender(&sender_id); + } +} + +#[cfg(test)] +mod sessions { + use super::*; + use crate::{ + data_director::{Include, Path, RegMsg, Register}, + document::{clock::Clock, create::CreateDoc, field::Field}, + message::{Addition, MsgAction, Query, Records}, + mtterror::MTTError, + name::{Name, NameType, Names}, + support_tests::TIMEOUT, + }; + use chrono::{DateTime, Utc}; + use std::{ + collections::HashSet, + sync::mpsc::{Receiver, RecvTimeoutError}, + thread::sleep, + }; + use uuid::Uuid; + + struct Setup { + queue: Queue, + rx: Receiver, + sender_id: Uuid, + } + + impl Setup { + fn new() -> Self { + let mut queue = Queue::new(); + let (tx, rx) = channel(); + let id = queue.add_sender(tx); + CreateDoc::start(queue.clone()); + Clock::start(queue.clone()); + Session::start(queue.clone()); + let paths = [ + Path::new( + Include::All, + Include::Just(NameType::Name(Session::doc_names()[0].clone())), + Include::Just(Action::Records), + ), + Path::new( + Include::All, + Include::Just(NameType::Name(Session::doc_names()[0].clone())), + Include::Just(Action::Error), + ), + ]; + for path in paths.iter() { + let reg = Register::new(id.clone(), RegMsg::AddRoute(path.clone())); + queue.send(Message::new(NameType::None, reg)).unwrap(); + rx.recv_timeout(TIMEOUT).unwrap(); + } + Self { + queue: queue, + rx: rx, + sender_id: id, + } + } + + fn message(action: A) -> Message + where + A: Into, + { + Message::new(Session::doc_names()[0].clone(), action) + } + + fn recv(&self) -> Result { + self.rx.recv_timeout(TIMEOUT) + } + + fn send(&self, msg: Message) { + self.queue.send(msg).unwrap(); + } + + fn send_registry_message(&self, msg: RegMsg) { + let request = Register::new(self.sender_id.clone(), msg); + self.queue + .send(Message::new(NameType::None, request)) + .unwrap(); + } + } + + #[test] + fn is_session_document_created() { + let setup = Setup::new(); + for name in Session::doc_names().iter() { + let path = Path::new(Include::All, Include::Just(name.into()), Include::All); + setup.send_registry_message(RegMsg::AddRoute(path)); + let result = setup.recv().unwrap(); + let action = result.get_action(); + match action { + MsgAction::Register(data) => match data.get_msg() { + RegMsg::RouteID(_) => {} + _ => unreachable!("got {:?} should have been route id", data), + }, + _ => unreachable!("got {:?} should have been data register", action), + } + } + } + + #[test] + fn are_session_ids_unique() { + let setup = Setup::new(); + let mut ids: HashSet = HashSet::new(); + let count = 10; + for _ in 0..count { + let msg = Setup::message(Addition::new()); + setup.send(msg); + let result = setup.recv().unwrap(); + match result.get_action() { + MsgAction::Records(recs) => { + assert_eq!(recs.len(), 1); + let rec = recs.iter().last().unwrap(); + let holder = rec.get(Name::english("id")).unwrap(); + let id = match holder { + Field::Uuid(data) => data, + _ => unreachable!("got {:?} should have been uuid", holder), + }; + ids.insert(id); + } + _ => unreachable!("should always return records"), + } + } + assert_eq!(ids.len(), count, "should be {} ids, got {:?}", count, ids); + } + + #[test] + fn expire_default_is_an_hour_from_now() { + let setup = Setup::new(); + let msg = Setup::message(Addition::new()); + let start_time = Utc::now() + Duration::from_hours(1); + setup.send(msg); + let result = setup.recv().unwrap(); + let end_time = Utc::now() + Duration::from_hours(1); + let action = result.get_action(); + match action { + MsgAction::Records(recs) => { + assert_eq!(recs.len(), 1); + let rec = recs.iter().last().unwrap(); + let holder = rec.get(Name::english("expire")).unwrap(); + match holder { + Field::DateTime(data) => { + assert!(data > start_time, "expire should be after {:?}", start_time); + assert!(data < end_time, "expire should be before {:?}", end_time); + } + _ => unreachable!("got {:?} should have been date time", holder), + }; + } + _ => unreachable!("got {:?}, should have gotten records", action), + } + } + + #[test] + fn session_ids_error_when_not_unique() { + let setup = Setup::new(); + let id = Uuid::new_v4(); + let mut addition = Addition::new(); + addition.add_field(Name::english("id"), id); + setup.send(Setup::message(addition.clone())); + setup.recv().unwrap(); + setup.send(Setup::message(addition)); + let result = setup.recv().unwrap(); + let action = result.get_action(); + match action { + MsgAction::Error(err) => match err { + MTTError::FieldDuplicate => {} + _ => unreachable!("got {:?}, should have been a field duplicate", err), + }, + _ => unreachable!("got {:?}, should have been an error", action), + } + } + + #[test] + fn expire_should_update_on_successful_query() { + let setup = Setup::new(); + let id = Uuid::new_v4(); + let timestamp = Utc::now() + Duration::from_secs(60); + let mut addition = Addition::new(); + addition.add_field(Name::english("id"), id.clone()); + addition.add_field(Name::english("expire"), timestamp); + setup.send(Setup::message(addition)); + setup.recv().unwrap(); + let mut query = Query::new(); + let mut calc = Calculation::new(Operand::Equal); + calc.add_value(CalcValue::Existing(FieldType::Uuid)) + .unwrap(); + calc.add_value(id).unwrap(); + query.add(Name::english("id"), calc.clone()); + let get_expire_datetime = || -> DateTime { + setup.send(Setup::message(query.clone())); + let result = setup.recv().unwrap(); + let action = result.get_action(); + match action { + MsgAction::Records(recs) => { + assert_eq!(recs.len(), 1); + let rec = recs.iter().last().unwrap(); + let holder = rec.get(Name::english("expire")).unwrap(); + match holder { + Field::DateTime(data) => data.clone(), + _ => unreachable!("got {:?} should have been date time", holder), + } + } + _ => unreachable!("got {:?}, should have gotten records", action), + } + }; + let start_time = Utc::now() + Duration::from_secs(3600); + let first_query = get_expire_datetime(); + let end_time = Utc::now() + Duration::from_secs(3601); + assert_eq!(first_query, timestamp); + let second_query = get_expire_datetime(); + assert!( + second_query > start_time, + "{:?} should be after {:?}", + second_query, + start_time + ); + assert!( + second_query < end_time, + "{:?} should be before {:?}", + second_query, + end_time + ); + } + + #[test] + fn clock_removes_expired_sessions() { + let setup = Setup::new(); + let id1 = Uuid::new_v4(); + let id2 = Uuid::new_v4(); + let duration = Duration::from_secs(60); + let expire1 = Utc::now() + duration; + let expire2 = Utc::now() - duration; + let mut addition1 = Addition::new(); + addition1.add_field(Name::english("id"), id1.clone()); + addition1.add_field(Name::english("expire"), expire1); + let mut addition2 = Addition::new(); + addition2.add_field(Name::english("id"), id2); + addition2.add_field(Name::english("expire"), expire2); + setup.send(Setup::message(addition1)); + setup.send(Setup::message(addition2)); + setup.recv().unwrap(); // Eat addition result. + setup.recv().unwrap(); // Eat addition result. + setup.send(Message::new( + Name::english("clock"), + MsgAction::OnUpdate(Records::new(Names::new())), + )); + sleep(TIMEOUT); // Allow time to react to message. + setup.send(Setup::message(Query::new())); + let result = setup.recv().unwrap(); + let action = result.get_action(); + match action { + MsgAction::Records(recs) => { + assert_eq!(recs.len(), 1, "nothing was deleted"); + let rec = recs.iter().last().unwrap(); + let id = rec.get(Name::english("id")).unwrap(); + let expire = rec.get(Name::english("expire")).unwrap(); + assert_eq!(id, id1.into(), "\n\n{:?}\n{:?}", Utc::now(), recs); + assert_eq!(expire, expire1.into(), "\n\n{:?}\n{:?}", Utc::now(), recs); + } + _ => unreachable!("got {:?}, should have gotten records", action), + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 02ef908..ddcaf2c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,8 +10,9 @@ use document::{ clock::Clock, create::CreateDoc, field::{Field, FieldType}, + session::Session, }; -use message::{Action, Addition, CalcValue, Calculation, Message, Operand, Session}; +use message::{Action, Addition, CalcValue, Calculation, Message, Operand}; pub use message::{MsgAction, Query}; use name::{Name, NameType}; use router::Queue; @@ -35,8 +36,7 @@ impl MoreThanText { let queue = Queue::new(); Clock::start(queue.clone()); CreateDoc::start(queue.clone()); - let session = Session::new(); - session.create(queue.clone()); + Session::start(queue.clone()); Self { queue: queue } } diff --git a/src/message.rs b/src/message.rs index 1bcfa60..b6f770d 100644 --- a/src/message.rs +++ b/src/message.rs @@ -1,12 +1,11 @@ use crate::{ data_director::{Include, Path, RegMsg, Register, Route}, document::{ - create::IndexType, - definition::{DocDef, DocFuncType}, + definition::DocDef, field::{Field, FieldType}, }, mtterror::MTTError, - name::{Name, NameType, Names}, + name::{NameType, Names}, router::Queue, }; use chrono::prelude::*; @@ -159,6 +158,7 @@ impl From<&Uuid> for MsgAction { #[cfg(test)] mod msgactions { use super::*; + use crate::name::Name; #[test] fn turn_document_definition_into_action() { @@ -291,6 +291,7 @@ impl Message { #[cfg(test)] mod messages { use super::*; + use crate::name::Name; #[test] fn can_the_document_be_a_named_reference() { @@ -480,6 +481,7 @@ impl Addition { #[cfg(test)] mod additions { use super::*; + use crate::name::Name; #[test] fn can_add_static_string() { @@ -1252,6 +1254,7 @@ impl Query { #[cfg(test)] mod queries { use super::*; + use crate::name::Name; #[test] fn holds_calculation_to_run_query() { @@ -1314,6 +1317,7 @@ impl Reply { #[cfg(test)] mod replies { use super::*; + use crate::name::Name; #[test] fn is_new_empty() { @@ -1580,6 +1584,7 @@ impl Document { #[cfg(test)] mod documents { use super::*; + use crate::name::Name; #[test] fn can_add_static_string() { @@ -1719,6 +1724,7 @@ impl MsgEntry { #[cfg(test)] mod msg_entries { use super::*; + use crate::name::Name; #[test] fn creates_message_entry() { @@ -1773,6 +1779,7 @@ impl MsgLogs { #[cfg(test)] mod msg_logs { use super::*; + use crate::name::Name; #[test] fn can_add_message_to_log() { @@ -1878,7 +1885,7 @@ impl MessageLog { #[cfg(test)] mod message_logs { use super::*; - use crate::support_tests::TIMEOUT; + use crate::{name::Name, support_tests::TIMEOUT}; #[test] fn does_log_store_messages() { @@ -1921,355 +1928,3 @@ mod message_logs { } } } - -pub struct Session { - doc_name: Name, -} - -impl Session { - pub fn new() -> Self { - Self { - doc_name: Name::english("session"), - } - } - - #[allow(dead_code)] - pub fn get_document_name(&self) -> &Name { - &self.doc_name - } - - pub fn create(&self, mut queue: Queue) { - let mut docdef = DocDef::new(self.doc_name.clone()); - - let mut calc = Calculation::new(Operand::Add); - calc.add_value(FieldType::DateTime).unwrap(); - calc.add_value(Duration::from_hours(1)).unwrap(); - - let name_id = Name::english("id"); - docdef.add_field(name_id.clone(), FieldType::Uuid); - docdef.set_default(&name_id, FieldType::Uuid).unwrap(); - docdef.add_index(&name_id, IndexType::Unique).unwrap(); - - let name_expire = Name::english("expire"); - docdef.add_field(name_expire.clone(), FieldType::DateTime); - docdef.set_default(&name_expire, calc.clone()).unwrap(); - - let mut update = Update::new(Query::new()); - update - .get_values_mut() - .add_field(name_expire.clone(), calc.clone()); - let path = Path::new( - Include::All, - Include::Just(self.doc_name.clone().into()), - Include::Just(Action::OnQuery), - ); - let query_action = DocFuncType::ExistingQuery(update.into()); - docdef.add_route(path, query_action); - - let mut delete_qry = Query::new(); - let mut delete_calc = Calculation::new(Operand::LessThan); - delete_calc.add_value(FieldType::DateTime).unwrap(); - delete_calc - .add_value(CalcValue::Existing(FieldType::DateTime)) - .unwrap(); - delete_qry.add(name_expire.clone(), delete_calc); - let delete = Delete::new(delete_qry); - let clock_path = Path::new( - Include::All, - Include::Just(Name::english("clock").into()), - Include::Just(Action::OnUpdate), - ); - let delete_func = DocFuncType::Trigger(delete.into()); - docdef.add_route(clock_path, delete_func); - - let (tx, rx) = channel(); - let sender_id = queue.add_sender(tx); - let msg = Message::new(NameType::None, docdef.clone()); - let path = Path::new( - Include::Just(msg.get_message_id().clone()), - Include::All, - Include::All, - ); - let reg_msg = Register::new(sender_id.clone(), RegMsg::AddRoute(path)); - queue.send(msg.forward(NameType::None, reg_msg)).unwrap(); - rx.recv().unwrap(); // Wait for completion. - queue.send(msg).unwrap(); - rx.recv().unwrap(); // Wait for completion. - rx.recv().unwrap(); // Wait for completion. - queue.remove_sender(&sender_id); - } -} - -#[cfg(test)] -mod sessions { - use super::*; - use crate::{document::create::CreateDoc, support_tests::TIMEOUT}; - use std::{sync::mpsc::RecvTimeoutError, thread::sleep}; - - struct Setup { - queue: Queue, - rx: Receiver, - sender_id: Uuid, - } - - impl Setup { - fn new() -> Self { - let (tx, rx) = channel(); - let mut queue = Queue::new(); - let id = queue.add_sender(tx); - CreateDoc::start(queue.clone()); - crate::document::clock::Clock::start(queue.clone()); - Self { - queue: queue, - rx: rx, - sender_id: id, - } - } - - fn get_sender_id(&self) -> Uuid { - self.sender_id.clone() - } - - fn get_queue(&self) -> Queue { - self.queue.clone() - } - - fn recv(&self) -> Result { - self.rx.recv_timeout(TIMEOUT) - } - - fn register(&self) { - let session = Session::new(); - let paths = [ - Path::new( - Include::All, - Include::Just(session.get_document_name().into()), - Include::Just(Action::Error), - ), - Path::new( - Include::All, - Include::Just(session.get_document_name().into()), - Include::Just(Action::Records), - ), - ]; - for path in paths.iter() { - let reg_msg = Register::new(self.sender_id.clone(), RegMsg::AddRoute(path.clone())); - self.queue - .send(Message::new(NameType::None, reg_msg)) - .unwrap(); - self.rx.recv().unwrap(); // Wait for completion. - } - } - } - - #[test] - fn creates_the_session_table() { - let setup = Setup::new(); - let queue = setup.get_queue(); - let session = Session::new(); - assert_eq!(session.get_document_name(), &Name::english("session")); - session.create(queue.clone()); - let path = Path::new( - Include::All, - Include::Just(session.get_document_name().into()), - Include::All, - ); - let reg_msg = Register::new(setup.get_sender_id(), RegMsg::AddRoute(path)); - queue.send(Message::new(NameType::None, reg_msg)).unwrap(); - setup.recv().unwrap(); - } - - #[test] - fn session_ids_are_unique() { - let setup = Setup::new(); - let queue = setup.get_queue(); - let session = Session::new(); - session.create(queue.clone()); - setup.register(); - let count = 10; - let msg = Message::new(session.get_document_name(), Addition::new()); - let mut ids: Vec = Vec::new(); - for _ in 0..count { - queue.send(msg.clone()).unwrap(); - let result = setup.recv().unwrap(); - let action = result.get_action(); - match action { - MsgAction::Records(recs) => { - assert_eq!(recs.len(), 1); - let rec = recs.iter().last().unwrap(); - let holder = rec.get(Name::english("id")).unwrap(); - let id = match holder { - Field::Uuid(data) => data, - _ => unreachable!("got {:?} should have been uuid", holder), - }; - assert!(!ids.contains(&id), "{} duplicated in {:?}", id, ids); - ids.push(id); - } - _ => unreachable!("got {:?}, should have gotten records", action), - } - } - } - - #[test] - fn expire_default_is_an_hour_from_now() { - let setup = Setup::new(); - let queue = setup.get_queue(); - let session = Session::new(); - session.create(queue.clone()); - setup.register(); - let msg = Message::new(session.get_document_name(), Addition::new()); - let start_time = Utc::now() + Duration::from_hours(1); - queue.send(msg).unwrap(); - let result = setup.recv().unwrap(); - let end_time = Utc::now() + Duration::from_hours(1); - let action = result.get_action(); - match action { - MsgAction::Records(recs) => { - assert_eq!(recs.len(), 1); - let rec = recs.iter().last().unwrap(); - let holder = rec.get(Name::english("expire")).unwrap(); - match holder { - Field::DateTime(data) => { - assert!(data > start_time, "expire should be after {:?}", start_time); - assert!(data < end_time, "expire should be before {:?}", end_time); - } - _ => unreachable!("got {:?} should have been date time", holder), - }; - } - _ => unreachable!("got {:?}, should have gotten records", action), - } - } - - #[test] - fn session_ids_error_when_not_unique() { - let setup = Setup::new(); - let queue = setup.get_queue(); - let session = Session::new(); - session.create(queue.clone()); - setup.register(); - let id = Uuid::new_v4(); - let mut addition = Addition::new(); - addition.add_field(Name::english("id"), id); - queue - .send(Message::new(session.get_document_name(), addition.clone())) - .unwrap(); - setup.recv().unwrap(); - queue - .send(Message::new(session.get_document_name(), addition)) - .unwrap(); - let result = setup.recv().unwrap(); - let action = result.get_action(); - match action { - MsgAction::Error(err) => match err { - MTTError::FieldDuplicate => {} - _ => unreachable!("got {:?}, should have been a field duplicate", err), - }, - _ => unreachable!("got {:?}, should have been an error", action), - } - } - - #[test] - fn expire_should_update_on_successful_query() { - let setup = Setup::new(); - let queue = setup.get_queue(); - let session = Session::new(); - session.create(queue.clone()); - setup.register(); - let id = Uuid::new_v4(); - let timestamp = Utc::now(); - let mut addition = Addition::new(); - addition.add_field(Name::english("id"), id.clone()); - addition.add_field(Name::english("expire"), timestamp); - queue - .send(Message::new(session.get_document_name(), addition.clone())) - .unwrap(); - setup.recv().unwrap(); - let mut query = Query::new(); - let mut calc = Calculation::new(Operand::Equal); - calc.add_value(CalcValue::Existing(FieldType::Uuid)) - .unwrap(); - calc.add_value(id).unwrap(); - query.add(Name::english("id"), calc.clone()); - queue - .send(Message::new(session.get_document_name(), query.clone())) - .unwrap(); - setup.recv().unwrap(); - let start_time = Utc::now() + Duration::from_secs(3600); - queue - .send(Message::new(session.get_document_name(), query.clone())) - .unwrap(); - let result = setup.recv().unwrap(); - let end_time = Utc::now() + Duration::from_hours(1); - sleep(TIMEOUT); - let action = result.get_action(); - match action { - MsgAction::Records(recs) => { - assert_eq!(recs.len(), 1); - let rec = recs.iter().last().unwrap(); - let holder = rec.get(Name::english("expire")).unwrap(); - match holder { - Field::DateTime(data) => { - assert!(data > start_time, "expire should be after {:?}", start_time); - assert!(data < end_time, "expire should be before {:?}", end_time); - } - _ => unreachable!("got {:?} should have been date time", holder), - }; - } - _ => unreachable!("got {:?}, should have gotten records", action), - } - } - - #[test] - fn clock_removes_expired_sessions() { - let setup = Setup::new(); - let queue = setup.get_queue(); - let session = Session::new(); - session.create(queue.clone()); - setup.register(); - let id1 = Uuid::new_v4(); - let id2 = Uuid::new_v4(); - let duration = Duration::from_secs(240); - let expire1 = Utc::now() + duration; - println!("{:?}", expire1); - println!("{:?}", Utc::now()); - let expire2 = Utc::now() - duration; - println!("{:?}", expire2); - let mut addition1 = Addition::new(); - addition1.add_field(Name::english("id"), id1.clone()); - addition1.add_field(Name::english("expire"), expire1); - let mut addition2 = Addition::new(); - addition2.add_field(Name::english("id"), id2); - addition2.add_field(Name::english("expire"), expire2); - queue - .send(Message::new(session.get_document_name(), addition1)) - .unwrap(); - queue - .send(Message::new(session.get_document_name(), addition2)) - .unwrap(); - setup.recv().unwrap(); - setup.recv().unwrap(); - queue - .send(Message::new( - Name::english("clock"), - MsgAction::OnUpdate(Records::new(Names::new())), - )) - .unwrap(); - sleep(TIMEOUT); - queue - .send(Message::new(session.get_document_name(), Query::new())) - .unwrap(); - let result = setup.recv().unwrap(); - let action = result.get_action(); - match action { - MsgAction::Records(recs) => { - assert_eq!(recs.len(), 1, "nothing was deleted"); - let rec = recs.iter().last().unwrap(); - let id = rec.get(Name::english("id")).unwrap(); - let expire = rec.get(Name::english("expire")).unwrap(); - assert_eq!(id, id1.into(), "\n\n{:?}\n{:?}", Utc::now(), recs); - assert_eq!(expire, expire1.into(), "\n\n{:?}\n{:?}", Utc::now(), recs); - } - _ => unreachable!("got {:?}, should have gotten records", action), - } - } -} diff --git a/src/router.rs b/src/router.rs index 9eda15d..a2587e5 100644 --- a/src/router.rs +++ b/src/router.rs @@ -47,7 +47,10 @@ impl Router { if id == &Uuid::nil() { return; } - self.senders.get(id).unwrap().send(msg).unwrap(); + match self.senders.get(id) { + Some(sender) => sender.send(msg).unwrap(), + None => {} + } } fn send(&self, msg: Message) { @@ -100,34 +103,70 @@ mod routers { name::Name, support_tests::TIMEOUT, }; - use std::collections::HashSet; + use std::{ + collections::HashSet, + sync::mpsc::{Receiver, RecvTimeoutError}, + }; + + struct Setup { + test_mod: Router, + rx: Receiver, + } + + impl Setup { + fn new() -> Self { + let (tx, rx) = channel(); + Self { + test_mod: Router::new(tx), + rx: rx, + } + } + + fn get_router(&self) -> &Router { + &self.test_mod + } + + fn get_router_mut(&mut self) -> &mut Router { + &mut self.test_mod + } + + fn recv(&self) -> Result { + self.rx.recv_timeout(TIMEOUT) + } + } #[test] fn can_pass_message() { - let (tx, rx) = channel(); - let router = Router::new(tx); + let setup = Setup::new(); + let router = setup.get_router(); let msg = Message::new(Name::english("task"), Query::new()); router.send(msg.clone()); - let result = rx.recv_timeout(TIMEOUT).unwrap(); + let result = setup.recv().unwrap(); assert_eq!(result.get_message_id(), msg.get_message_id()); } #[test] fn can_forward_message() { - let (tx, _) = channel(); - let mut router = Router::new(tx); - let (sender, receiver) = channel(); - let id = router.add_sender(sender); - let msg = Message::new(Name::english("wiki"), Query::new()); - router.forward(&id, msg.clone()); - let result = receiver.recv_timeout(TIMEOUT).unwrap(); - assert_eq!(result.get_message_id(), msg.get_message_id()); + let mut setup = Setup::new(); + let router = setup.get_router_mut(); + let mut receivers: HashMap> = HashMap::new(); + for _ in 0..10 { + let (tx, rx) = channel(); + let id = router.add_sender(tx); + receivers.insert(id, rx); + } + for (id, recv) in receivers.iter() { + let msg = Message::new(Name::english(id.to_string().as_str()), Query::new()); + router.forward(id, msg.clone()); + let result = recv.recv_timeout(TIMEOUT).unwrap(); + assert_eq!(result.get_message_id(), msg.get_message_id()); + } } #[test] fn sender_ids_are_unique() { - let (tx, _) = channel(); - let mut router = Router::new(tx); + let mut setup = Setup::new(); + let router = setup.get_router_mut(); let count = 10; let mut holder: HashSet = HashSet::new(); for _ in 0..count { @@ -139,33 +178,154 @@ mod routers { #[test] fn can_remove_sender() { - let (tx, rx) = channel(); - let mut router = Router::new(tx); - let (data, _) = channel(); - let id = router.add_sender(data); - assert_eq!(router.senders.len(), 1, "should have only one sender"); - router.remove_sender(&id); - assert_eq!(router.senders.len(), 0, "should have no senders."); - let result = rx.recv_timeout(TIMEOUT).unwrap(); - let action = result.get_action(); + let mut setup = Setup::new(); + let router = setup.get_router_mut(); + let mut receivers: HashMap> = HashMap::new(); + for _ in 0..10 { + let (tx, rx) = channel(); + let id = router.add_sender(tx); + receivers.insert(id, rx); + } + let removed = receivers.keys().last().unwrap().clone(); + router.remove_sender(&removed); + let router = setup.get_router(); + let removed_recv = receivers.remove(&removed).unwrap(); + router.forward(&removed, Message::new(NameType::None, Query::new())); + match removed_recv.recv_timeout(TIMEOUT) { + Err(err) => match err { + RecvTimeoutError::Disconnected => {} + _ => unreachable!("got {:?}, should have been disconnected", err), + }, + _ => unreachable!("should have returned an error"), + } + let announce = setup.recv().unwrap(); + let action = announce.get_action(); match action { - MsgAction::Register(reg_msg) => { - let reg_action = reg_msg.get_msg(); - match reg_action { - RegMsg::RemoveSender(result) => assert_eq!(result, &id), - _ => unreachable!("got {:?}, should have been remove sender", reg_action), + MsgAction::Register(data) => { + let output = data.get_msg(); + match output { + RegMsg::RemoveSender(id) => assert_eq!(id, &removed), + _ => unreachable!("got {:?} should have been sender removal", output), } } - _ => unreachable!("got {:?}, should have been registry message", action), + _ => unreachable!("got {:?}, should have been register", action), + } + for (id, recv) in receivers.iter() { + let msg = Message::new(Name::english(id.to_string().as_str()), Query::new()); + router.forward(id, msg.clone()); + let result = recv.recv_timeout(TIMEOUT).unwrap(); + assert_eq!(result.get_message_id(), msg.get_message_id()); } } #[test] fn ignores_bad_id_removals() { - let (tx, rx) = channel(); - let mut router = Router::new(tx); - router.remove_sender(&Uuid::new_v4()); + let mut setup = Setup::new(); + let router = setup.get_router_mut(); + let removed = Uuid::new_v4(); + router.remove_sender(&removed); assert_eq!(router.senders.len(), 0, "should have no senders."); - rx.recv_timeout(TIMEOUT).unwrap(); + let announce = setup.recv().unwrap(); + let action = announce.get_action(); + match action { + MsgAction::Register(data) => { + let output = data.get_msg(); + match output { + RegMsg::RemoveSender(id) => assert_eq!(id, &removed), + _ => unreachable!("got {:?} should have been sender removal", output), + } + } + _ => unreachable!("got {:?}, should have been register", action), + } + } +} + +#[cfg(test)] +mod queues { + use super::*; + use crate::{ + data_director::{Include, Path}, + message::MsgAction, + name::Name, + support_tests::TIMEOUT, + }; + use std::sync::mpsc::{Receiver, RecvTimeoutError}; + + struct Setup { + test_mod: Queue, + rx: Receiver, + rx_id: Uuid, + } + + impl Setup { + fn new() -> Self { + let mut queue = Queue::new(); + let (tx, rx) = channel(); + let id = queue.add_sender(tx); + Self { + test_mod: queue, + rx: rx, + rx_id: id, + } + } + + fn send_reg_msg(&self, msg: RegMsg) { + let reg_msg = Register::new(self.rx_id.clone(), msg); + self.test_mod + .send(Message::new(NameType::None, reg_msg)) + .unwrap(); + } + + fn recv(&self) -> Result { + self.rx.recv_timeout(TIMEOUT) + } + } + + #[test] + fn can_add_names_registry() { + let setup = Setup::new(); + let name = Name::english(Uuid::new_v4().to_string().as_str()); + let reg = RegMsg::AddDocName(vec![name.clone()]); + setup.send_reg_msg(reg); + let result = setup.recv().unwrap(); + let action = result.get_action(); + match action { + MsgAction::Register(data) => { + let regmsg = data.get_msg(); + match data.get_msg() { + RegMsg::DocumentNameID(_) => {} + _ => unreachable!("got {:?} should have been document id", regmsg), + } + } + _ => unreachable!("got {:?} should have been register", action), + } + } + + #[test] + fn returns_error_when_document_name_not_found() { + let setup = Setup::new(); + let name = Name::english(Uuid::new_v4().to_string().as_str()); + let path = Path::new( + Include::All, + Include::Just(name.clone().into()), + Include::All, + ); + let reg = RegMsg::AddRoute(path); + setup.send_reg_msg(reg); + let result = setup.recv().unwrap(); + let action = result.get_action(); + match action { + MsgAction::Register(data) => { + let regmsg = data.get_msg(); + match data.get_msg() { + RegMsg::Error(err) => match err { + MTTError::NameNotFound(failed_name) => assert_eq!(failed_name, &name), + _ => unreachable!("got {:?} should have been missing name", err), + }, + _ => unreachable!("got {:?} should have been error", regmsg), + } + } + _ => unreachable!("got {:?} should have been register", action), + } } }