Got session control into it's own layer.

This commit is contained in:
Jeff Baskin 2025-04-21 21:44:52 -04:00
parent 4fc050e590
commit 393b66a9f5
6 changed files with 168 additions and 369 deletions

View File

@ -15,373 +15,165 @@ use uuid::Uuid;
const RESPONS_TO: [MsgType; 2] = [MsgType::Document, MsgType::SessionValidated]; const RESPONS_TO: [MsgType; 2] = [MsgType::Document, MsgType::SessionValidated];
pub struct Request {
pub session: Option<Field>,
}
impl Request {
pub fn new(session: Option<Field>) -> Self {
Self { session: session }
}
}
#[cfg(test)]
pub mod requests {
use super::*;
pub fn get_root_document() -> Request {
Request::new(None)
}
pub fn get_root_with_session(sess_id: &Uuid) -> Request {
Request::new(Some(sess_id.clone().into()))
}
pub fn get_root_document_eith_session<F>(id: F) -> Request
where
F: Into<Field>,
{
Request::new(Some(id.into()))
}
#[test]
fn new_request_no_session() {
let sess: Option<Field> = None;
let req = Request::new(sess);
assert!(req.session.is_none(), "should not have a session")
}
#[test]
fn new_request_with_session() {
let id = Uuid::new_v4();
let req = Request::new(Some(id.into()));
match req.session {
Some(result) => assert_eq!(result.to_uuid().unwrap(), id),
None => unreachable!("should contain a session"),
}
}
}
pub struct Reply {
sess_id: Uuid,
content: String,
}
impl Reply {
fn new(sess_id: Uuid, content: String) -> Self {
Self {
sess_id: sess_id,
content: content,
}
}
pub fn get_session(&self) -> Uuid {
self.sess_id.clone()
}
pub fn get_content(&self) -> String {
self.content.clone()
}
}
#[cfg(test)]
mod replies {
use super::*;
pub fn create_reply() -> Reply {
Reply {
sess_id: Uuid::new_v4(),
content: "some text".to_string(),
}
}
#[test]
fn create_new_reply() {
let sess_id = Uuid::new_v4();
let txt = Uuid::new_v4().to_string();
let reply = Reply::new(sess_id, txt.clone());
assert_eq!(reply.get_session(), sess_id);
assert_eq!(reply.get_content(), txt);
}
}
#[derive(Clone)] #[derive(Clone)]
pub struct ClientRegistry { pub struct ClientChannel {
queue: Queue,
registry: Arc<Mutex<HashMap<Uuid, Sender<Message>>>>, registry: Arc<Mutex<HashMap<Uuid, Sender<Message>>>>,
} }
impl ClientRegistry { impl ClientChannel {
pub fn new() -> Self { fn new(queue: Queue) -> Self {
Self { Self {
queue: queue,
registry: Arc::new(Mutex::new(HashMap::new())), registry: Arc::new(Mutex::new(HashMap::new())),
} }
} }
fn get_id<'a>( pub fn send(&self, mut msg: Message) -> Receiver<Message> {
gen: &mut impl Iterator<Item = Uuid>,
data: &HashMap<Uuid, Sender<Message>>,
) -> Uuid {
let mut id = gen.next().unwrap();
while data.contains_key(&id) {
id = gen.next().unwrap();
}
id.clone()
}
pub fn add(&mut self, tx: Sender<Message>) -> Uuid {
let mut reg = self.registry.lock().unwrap(); let mut reg = self.registry.lock().unwrap();
let mut gen_id = GenID::new(); if reg.contains_key(&msg.get_id()) {
let id = ClientRegistry::get_id(&mut gen_id, &reg); let mut id = Uuid::new_v4();
reg.insert(id.clone(), tx); while reg.contains_key(&id) {
id id = Uuid::new_v4();
} }
msg.reset_id(id);
fn send(&mut self, id: &Uuid, msg: Message) {
let mut reg = self.registry.lock().unwrap();
let tx = reg.remove(id).unwrap();
tx.send(msg).unwrap();
}
}
#[cfg(test)]
mod clientregistries {
use super::*;
use crate::client::replies::create_reply;
use std::{
sync::mpsc::{channel, Receiver},
time::Duration,
};
static TIMEOUT: Duration = Duration::from_millis(500);
#[test]
fn create_client_registry() {
let reg = ClientRegistry::new();
let data = reg.registry.lock().unwrap();
assert!(data.is_empty(), "needs to create an empty hashmap");
}
#[test]
fn send_from_client() {
let mut reg = ClientRegistry::new();
let count = 10;
let mut rxs: HashMap<Uuid, Receiver<Message>> = HashMap::new();
for _ in 0..count {
let (tx, rx) = channel::<Message>();
let id = reg.add(tx);
rxs.insert(id, rx);
} }
assert_eq!(rxs.len(), count, "should have been {} receivers", count);
for (id, rx) in rxs.iter() {
let msg = Message::new(MsgType::Document);
reg.send(id, msg);
rx.recv_timeout(TIMEOUT).unwrap();
}
let data = reg.registry.lock().unwrap();
assert!(data.is_empty(), "should remove sender after sending");
}
#[test]
fn prevent_duplicates() {
let mut reg = ClientRegistry::new();
let (tx, _rx) = channel::<Message>();
let existing = reg.add(tx);
let expected = Uuid::new_v4();
let ids = [existing.clone(), expected.clone()];
let data = reg.registry.lock().unwrap();
let result = ClientRegistry::get_id(&mut ids.into_iter(), &data);
assert_eq!(result, expected);
}
}
#[derive(Clone)]
pub struct ClientLink {
tx: Sender<Message>,
registry: ClientRegistry,
}
impl ClientLink {
fn new(tx: Sender<Message>, registry: ClientRegistry) -> Self {
Self {
tx: tx,
registry: registry,
}
}
pub fn send(&mut self, mut req: Message) -> Receiver<Message> {
let (tx, rx) = channel(); let (tx, rx) = channel();
//let mut msg: Message = req.into(); reg.insert(msg.get_id(), tx);
let id = self.registry.add(tx); self.queue.send(msg);
req.add_data("tx_id", id);
self.tx.send(req).unwrap();
rx rx
} }
fn reply(&self, msg: Message) {
let mut reg = self.registry.lock().unwrap();
match reg.remove(&msg.get_id()) {
Some(tx) => tx.send(msg).unwrap(),
None => {}
}
}
} }
#[cfg(test)] #[cfg(test)]
mod clientlinks { mod client_channels {
use super::*; use super::*;
use crate::client::replies::create_reply;
use std::time::Duration; use std::time::Duration;
static TIMEOUT: Duration = Duration::from_millis(500); static TIMEOUT: Duration = Duration::from_millis(500);
#[test] #[test]
fn create_client_link() { fn request_new_message() {
let queue = Queue::new();
let chan = ClientChannel::new(queue);
let msg_types = [MsgType::Document, MsgType::Time];
for msg_type in msg_types.iter() {
let msg = Message::new(msg_type.clone());
assert_eq!(msg.get_msg_type(), msg_type);
}
}
#[test]
fn fowards_message() {
let msg_type = MsgType::Document;
let reply_type = MsgType::Time;
let queue = Queue::new();
let (tx, rx) = channel(); let (tx, rx) = channel();
let mut registry = ClientRegistry::new(); queue.add(tx, [msg_type.clone()].to_vec());
let mut link = ClientLink::new(tx, registry.clone()); let chan = ClientChannel::new(queue);
let req = Request::new(None); let msg = Message::new(msg_type.clone());
let rx_client = link.send(req.into()); let client_rx = chan.send(msg.clone());
let msg = rx.recv_timeout(TIMEOUT).unwrap(); let reply = rx.recv_timeout(TIMEOUT).unwrap();
match msg.get_msg_type() { assert_eq!(reply.get_id(), msg.get_id());
MsgType::ClientRequest => {} assert_eq!(reply.get_msg_type().clone(), msg_type);
_ => unreachable!("should have been a client request"), let client_reply = reply.reply(MsgType::Time);
} chan.reply(client_reply);
match msg.get_data("tx_id") { let client_msg = client_rx.recv_timeout(TIMEOUT).unwrap();
Some(result) => { assert_eq!(client_msg.get_id(), msg.get_id());
let id = result.to_uuid().unwrap(); assert_eq!(client_msg.get_msg_type().clone(), reply_type);
registry.send(&id, msg.reply(MsgType::Document)); }
rx_client.recv().unwrap();
} #[test]
None => unreachable!("should have had a seender id"), fn no_duplicate_ids() {
} let (tx, rx) = channel();
let queue = Queue::new();
queue.add(tx, [MsgType::Time].to_vec());
let chan = ClientChannel::new(queue);
let msg1 = Message::new(MsgType::Time);
let msg2 = msg1.reply(MsgType::Time);
let rx1 = chan.send(msg1);
let rx2 = chan.send(msg2);
let queue1 = rx.recv_timeout(TIMEOUT).unwrap();
let queue2 = rx.recv_timeout(TIMEOUT).unwrap();
assert_ne!(queue1.get_id(), queue2.get_id());
chan.reply(queue1.reply(MsgType::Document));
chan.reply(queue2.reply(MsgType::Document));
let reply1 = rx1.recv_timeout(TIMEOUT).unwrap();
let reply2 = rx2.recv_timeout(TIMEOUT).unwrap();
assert_eq!(reply1.get_id(), queue1.get_id());
assert_eq!(reply2.get_id(), queue2.get_id());
}
#[test]
fn ignore_unrequested() {
let queue = Queue::new();
let chan = ClientChannel::new(queue);
chan.reply(Message::new(MsgType::Document));
} }
} }
pub struct Client { pub struct Client {
channel: ClientChannel,
queue: Queue, queue: Queue,
registry: ClientRegistry,
return_to: HashMap<Uuid, Message>,
rx: Receiver<Message>, rx: Receiver<Message>,
} }
impl Client { impl Client {
fn new(rx: Receiver<Message>, queue: Queue) -> Self { fn new(chan: ClientChannel, queue: Queue, rx: Receiver<Message>) -> Self {
Self { Self {
channel: chan,
queue: queue, queue: queue,
registry: ClientRegistry::new(),
return_to: HashMap::new(),
rx: rx, rx: rx,
} }
} }
pub fn start(queue: Queue) -> ClientLink { pub fn start(queue: Queue) -> ClientChannel {
let (tx, rx) = channel(); let (tx, rx) = channel();
queue.add(tx.clone(), RESPONS_TO.to_vec()); queue.add(tx.clone(), RESPONS_TO.to_vec());
let mut client = Client::new(rx, queue); let chan = ClientChannel::new(queue.clone());
let link = ClientLink::new(tx, client.get_registry()); let client = Client::new(chan.clone(), queue, rx);
spawn(move || { spawn(move || {
client.listen(); client.listen();
}); });
link chan
} }
fn listen(&mut self) { fn listen(&self) {
loop { loop {
let msg = self.rx.recv().unwrap(); let msg = self.rx.recv().unwrap();
match msg.get_msg_type() { self.channel.reply(msg);
MsgType::ClientRequest => self.client_request(msg),
MsgType::Document => self.document(msg),
MsgType::SessionValidated => self.session(msg),
_ => unreachable!("Received message it did not understand"),
}
} }
} }
fn get_registry(&self) -> ClientRegistry {
self.registry.clone()
}
fn client_request(&mut self, msg: Message) {
self.return_to.insert(msg.get_id(), msg.clone());
let mut reply = msg.reply(MsgType::SessionValidate);
match msg.get_data("sess_id") {
Some(sess_id) => reply.add_data("sess_id", sess_id.clone()),
None => {}
}
self.queue.send(reply).unwrap();
}
fn session(&mut self, msg: Message) {
let initial_msg = self.return_to.get_mut(&msg.get_id()).unwrap();
let mut reply = msg.reply(MsgType::DocumentRequest);
match msg.get_data("sess_id") {
Some(sess_id) => {
initial_msg.add_data("sess_id", sess_id.clone());
reply.add_data("sess_id", sess_id.clone());
}
None => unreachable!("validated should always have an id"),
}
self.queue.send(reply).unwrap();
}
fn document(&mut self, msg: Message) {
let initial_msg = self.return_to.remove(&msg.get_id()).unwrap();
let tx_id = initial_msg.get_data("tx_id").unwrap().to_uuid().unwrap();
/*
let reply = Reply::new(
initial_msg.get_data("sess_id").unwrap().to_uuid().unwrap(),
msg.get_data("doc").unwrap().to_string(),
)s
*/
self.registry
.send(&tx_id, initial_msg.reply(MsgType::Document));
}
} }
#[cfg(test)] #[cfg(test)]
mod clients { mod clients {
use super::*; use super::*;
use requests::get_root_with_session; use crate::session::sessions::create_validated_reply;
use std::time::Duration; use std::time::Duration;
static TIMEOUT: Duration = Duration::from_millis(500); static TIMEOUT: Duration = Duration::from_millis(500);
/*
#[test] #[test]
fn start_client() { fn session_validated() {
let sess_id1 = Uuid::new_v4();
let sess_id2 = Uuid::new_v4();
let doc = Uuid::new_v4().to_string();
let (tx, rx) = channel();
let queue = Queue::new(); let queue = Queue::new();
queue.add( let (queue_tx, queue_rx) = channel();
tx, queue.add(queue_tx, [MsgType::SessionValidate].to_vec());
[MsgType::SessionValidate, MsgType::DocumentRequest].to_vec(), let chan = Client::start(queue.clone());
); let chan_rx = chan.send(Message::new(MsgType::SessionValidate));
let mut link = Client::start(queue.clone()); let msg = queue_rx.recv_timeout(TIMEOUT).unwrap();
let req = get_root_with_session(&sess_id1); let expected = create_validated_reply(msg);
let reply_rx = link.send(req.into()); queue.send(expected.clone());
let send1 = rx.recv_timeout(TIMEOUT).unwrap(); let result = chan_rx.recv_timeout(TIMEOUT).unwrap();
match send1.get_msg_type() { assert_eq!(result.get_id(), expected.get_id());
MsgType::SessionValidate => {} assert_eq!(result.get_msg_type(), expected.get_msg_type());
_ => unreachable!("should request session validation"),
}
assert_eq!(
send1.get_data("sess_id").unwrap().to_uuid().unwrap(),
sess_id1
);
assert!(send1.get_data("tx_id").is_none());
let mut response = send1.reply_with_data(MsgType::SessionValidated);
response.add_data("sess_id", sess_id2);
queue.send(response).unwrap();
let send2 = rx.recv_timeout(TIMEOUT).unwrap();
assert_eq!(send2.get_id(), send1.get_id());
match send2.get_msg_type() {
MsgType::DocumentRequest => {}
_ => unreachable!("should request session validation"),
}
assert_eq!(
send2.get_data("sess_id").unwrap().to_uuid().unwrap(),
sess_id2
);
let mut document = send2.reply(MsgType::Document);
document.add_data("doc", doc.clone());
queue.send(document).unwrap();
let reply = reply_rx.recv_timeout(TIMEOUT).unwrap();
assert_eq!(reply.get_data("sess_id").unwrap().to_uuid().unwrap(), sess_id2);
assert_eq!(reply.get_data("doc").unwrap().to_string(), doc);
} }
*/
} }

View File

@ -43,13 +43,17 @@ impl Document {
} }
#[cfg(test)] #[cfg(test)]
mod documents { pub mod documents {
use super::*; use super::*;
use std::time::Duration; use std::time::Duration;
use uuid::Uuid; use uuid::Uuid;
const TIMEOUT: Duration = Duration::from_millis(500); const TIMEOUT: Duration = Duration::from_millis(500);
pub fn get_root_document() -> Message {
Message::new(MsgType::DocumentRequest)
}
fn setup_document(listen_for: Vec<MsgType>) -> (Queue, Receiver<Message>) { fn setup_document(listen_for: Vec<MsgType>) -> (Queue, Receiver<Message>) {
let queue = Queue::new(); let queue = Queue::new();
let (tx, rx) = channel(); let (tx, rx) = channel();

View File

@ -6,7 +6,7 @@ mod queue;
mod session; mod session;
mod utils; mod utils;
use client::{Client, ClientLink, Reply, Request}; use client::{Client, ClientChannel};
use clock::Clock; use clock::Clock;
use document::Document; use document::Document;
use field::Field; use field::Field;
@ -16,7 +16,7 @@ use uuid::Uuid;
#[derive(Clone)] #[derive(Clone)]
pub struct MoreThanText { pub struct MoreThanText {
client_link: ClientLink, client_channel: ClientChannel,
} }
impl MoreThanText { impl MoreThanText {
@ -26,7 +26,7 @@ impl MoreThanText {
Document::start(queue.clone()); Document::start(queue.clone());
Session::start(queue.clone()); Session::start(queue.clone());
Self { Self {
client_link: Client::start(queue.clone()), client_channel: Client::start(queue.clone()),
} }
} }
@ -39,7 +39,7 @@ impl MoreThanText {
Some(id) => msg.add_data("sess_id", id.into()), Some(id) => msg.add_data("sess_id", id.into()),
None => {} None => {}
} }
let rx = self.client_link.send(msg); let rx = self.client_channel.send(msg);
let reply = rx.recv().unwrap(); let reply = rx.recv().unwrap();
reply.get_data("sess_id").unwrap().to_uuid().unwrap() reply.get_data("sess_id").unwrap().to_uuid().unwrap()
} }
@ -52,8 +52,8 @@ impl MoreThanText {
Some(id) => Some(id.into()), Some(id) => Some(id.into()),
None => None, None => None,
}; };
let req = Request::new(sess); let req = Message::new(MsgType::DocumentRequest);
let rx = self.client_link.send(req.into()); let rx = self.client_channel.send(req.into());
rx.recv().unwrap() rx.recv().unwrap()
} }
} }

View File

@ -6,12 +6,11 @@ use axum::{
routing::get, routing::get,
RequestPartsExt, Router, RequestPartsExt, Router,
}; };
use axum_extra::extract::cookie::{Cookie, CookieJar};
use clap::Parser; use clap::Parser;
use morethantext::MoreThanText; use morethantext::MoreThanText;
use std::convert::Infallible; use std::convert::Infallible;
use tokio::{spawn, sync::mpsc::channel}; use tokio::{spawn, sync::mpsc::channel};
use tower_cookies::{CookieManagerLayer, Cookies}; use tower_cookies::{Cookie, CookieManagerLayer, Cookies};
use uuid::Uuid; use uuid::Uuid;
const LOCALHOST: &str = "127.0.0.1"; const LOCALHOST: &str = "127.0.0.1";
@ -64,34 +63,27 @@ where
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> { async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
let Extension(cookies) = parts.extract::<Extension<Cookies>>().await.unwrap(); let Extension(cookies) = parts.extract::<Extension<Cookies>>().await.unwrap();
let Extension(mut state) = parts.extract::<Extension<MoreThanText>>().await.unwrap(); let Extension(mut state) = parts.extract::<Extension<MoreThanText>>().await.unwrap();
let sess_id: Option<String> = None; let req_id = match cookies.get(SESSION_KEY) {
let id = Uuid::nil(); Some(cookie) => Some(cookie.value().to_string()),
//let id = state.validate_session(sess_id); None => None,
cookies.add(tower_cookies::Cookie::new(SESSION_KEY, id.to_string())); };
let requested = req_id.clone();
let (tx, mut rx) = channel(1);
spawn(async move {
tx.send(state.validate_session(requested)).await.unwrap();
});
let id = rx.recv().await.unwrap();
if !req_id.is_some_and(|x| x == id.to_string()) {
cookies.add(Cookie::new(SESSION_KEY, id.to_string()));
}
Ok(SessionID(id)) Ok(SessionID(id))
} }
} }
async fn mtt_conn( async fn mtt_conn(
jar: CookieJar,
sess_id: SessionID, sess_id: SessionID,
state: State<MoreThanText>, state: State<MoreThanText>,
) -> impl IntoResponse { ) -> impl IntoResponse {
/*
let sid = match jar.get(SESSION_KEY) {
Some(cookie) => Some(cookie.value().to_string()),
None => None,
};
let sess_info = sid.clone();
let (tx, mut rx) = channel(5);
spawn(async move {
tx.send(state.clone().request(sess_info)).await.unwrap();
});
let reply = rx.recv().await.unwrap();
let cookie = Cookie::build((SESSION_KEY, reply.get_data("sess_id").unwrap().to_string()));
let cookies = jar.add(cookie);
(cookies, reply.get_data("dov").unwrap().to_string())
*/
("something".to_string(),) ("something".to_string(),)
} }
@ -100,7 +92,10 @@ mod servers {
use super::*; use super::*;
use axum::{ use axum::{
body::Body, body::Body,
http::{Request, StatusCode}, http::{
header::{COOKIE, SET_COOKIE},
Request, StatusCode,
},
}; };
use tower::ServiceExt; use tower::ServiceExt;
@ -112,7 +107,7 @@ mod servers {
.await .await
.unwrap(); .unwrap();
assert_eq!(response.status(), StatusCode::OK); assert_eq!(response.status(), StatusCode::OK);
let sessid = format!("{:?}", response.headers().get("set-cookie").unwrap()); let sessid = format!("{:?}", response.headers().get(SET_COOKIE).unwrap());
assert!(sessid.contains(SESSION_KEY), "did not set session id"); assert!(sessid.contains(SESSION_KEY), "did not set session id");
} }
@ -126,7 +121,7 @@ mod servers {
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap()) .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await .await
.unwrap(); .unwrap();
let sessid = format!("{:?}", response.headers().get("set-cookie").unwrap()); let sessid = format!("{:?}", response.headers().get(SET_COOKIE).unwrap());
assert!( assert!(
!holder.contains(&sessid), !holder.contains(&sessid),
"found duplicate entry: {:?}", "found duplicate entry: {:?}",
@ -136,6 +131,29 @@ mod servers {
} }
} }
#[tokio::test]
async fn cookie_only_issued_once() {
let app = create_app(MoreThanText::new()).await;
let initial = app
.clone()
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(initial.status(), StatusCode::OK);
let sessid = initial.headers().get(SET_COOKIE).unwrap();
let mut request = Request::builder()
.uri("/")
.header(COOKIE, sessid.clone())
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
match response.headers().get(SET_COOKIE) {
Some(info) => assert!(false, "first pass: {:?}, second pass: {:?}", sessid, info),
None => {}
}
}
#[tokio::test] #[tokio::test]
async fn receive_file_not_found() { async fn receive_file_not_found() {
let app = create_app(MoreThanText::new()).await; let app = create_app(MoreThanText::new()).await;

View File

@ -1,4 +1,4 @@
use crate::{client::Request, field::Field}; use crate::field::Field;
use std::{ use std::{
collections::HashMap, collections::HashMap,
sync::{mpsc::Sender, Arc, RwLock}, sync::{mpsc::Sender, Arc, RwLock},
@ -70,23 +70,16 @@ impl Message {
pub fn get_id(&self) -> Uuid { pub fn get_id(&self) -> Uuid {
self.id.clone() self.id.clone()
} }
}
impl From<Request> for Message { pub fn reset_id(&mut self, id: Uuid) {
fn from(value: Request) -> Self { self.id = id;
let mut msg = Message::new(MsgType::ClientRequest);
match value.session {
Some(id) => msg.add_data("sess_id", id),
None => {}
}
msg
} }
} }
#[cfg(test)] #[cfg(test)]
mod messages { mod messages {
use super::*; use super::*;
use crate::client::requests::{get_root_document, get_root_document_eith_session}; use crate::document::documents::get_root_document;
#[test] #[test]
fn new_message() { fn new_message() {
@ -182,24 +175,10 @@ mod messages {
} }
#[test] #[test]
fn from_request_no_session() { fn reset_msg_id() {
let req = get_root_document(); let mut msg = Message::new(MsgType::Time);
let msg: Message = req.into(); msg.reset_id(Uuid::nil());
assert!( assert_eq!(msg.get_id(), Uuid::nil());
msg.get_data("sess_id").is_none(),
"should not have a session id"
)
}
#[test]
fn from_request_with_session() {
let id = Uuid::new_v4();
let req = get_root_document_eith_session(id.clone());
let msg: Message = req.into();
match msg.get_data("sess_id") {
Some(result) => assert_eq!(result.to_uuid().unwrap(), id),
None => unreachable!("should return an id"),
}
} }
} }

View File

@ -156,13 +156,19 @@ impl Session {
} }
#[cfg(test)] #[cfg(test)]
mod sessions { pub mod sessions {
use super::*; use super::*;
use crate::queue::{Message, MsgType}; use crate::queue::{Message, MsgType};
use std::{sync::mpsc::channel, time::Duration}; use std::{sync::mpsc::channel, time::Duration};
static TIMEOUT: Duration = Duration::from_millis(500); static TIMEOUT: Duration = Duration::from_millis(500);
pub fn create_validated_reply(msg: Message) -> Message {
let mut reply = msg.reply(MsgType::SessionValidated);
reply.add_data("sess_id", Uuid::new_v4());
reply
}
fn setup_session(listen_for: Vec<MsgType>) -> (Queue, Receiver<Message>) { fn setup_session(listen_for: Vec<MsgType>) -> (Queue, Receiver<Message>) {
let queue = Queue::new(); let queue = Queue::new();
let (tx, rx) = channel(); let (tx, rx) = channel();