MoreThanText/morethantext
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved session into own module.
Some checks failed
Gitea Actions Demo / Explore-Gitea-Actions (push) Failing after 1s
Details

Browse Source
This commit is contained in:
Jeff Baskin
2026-01-26 14:18:40 -05:00
parent 65bf323d3d
commit 102e60bf94
7 changed files with 563 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files

367
src/message.rs
View File

@@ -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<Message>,
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<Message, RecvTimeoutError> {
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<Uuid> = 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),
}
}
}