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

Preparing to move session into an extractor.

Browse Source
This commit is contained in:
Jeff Baskin
2025-04-19 07:57:16 -04:00
parent 9e6d407b69
commit bb70cc65e0
5 changed files with 146 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
src/client.rs
View File

@ -107,7 +107,7 @@ mod replies {
#[derive(Clone)]
pub struct ClientRegistry {
registry: Arc<Mutex<HashMap<Uuid, Sender<Reply>>>>,
registry: Arc<Mutex<HashMap<Uuid, Sender<Message>>>>,
}
impl ClientRegistry {
@ -119,7 +119,7 @@ impl ClientRegistry {
fn get_id<'a>(
gen: &mut impl Iterator<Item = Uuid>,
data: &HashMap<Uuid, Sender<Reply>>,
data: &HashMap<Uuid, Sender<Message>>,
) -> Uuid {
let mut id = gen.next().unwrap();
while data.contains_key(&id) {
@ -128,7 +128,7 @@ impl ClientRegistry {
id.clone()
}
pub fn add(&mut self, tx: Sender<Reply>) -> Uuid {
pub fn add(&mut self, tx: Sender<Message>) -> Uuid {
let mut reg = self.registry.lock().unwrap();
let mut gen_id = GenID::new();
let id = ClientRegistry::get_id(&mut gen_id, &reg);
@ -136,7 +136,7 @@ impl ClientRegistry {
id
}
fn send(&mut self, id: &Uuid, msg: Reply) {
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();
@ -165,15 +165,15 @@ mod clientregistries {
fn send_from_client() {
let mut reg = ClientRegistry::new();
let count = 10;
let mut rxs: HashMap<Uuid, Receiver<Reply>> = HashMap::new();
let mut rxs: HashMap<Uuid, Receiver<Message>> = HashMap::new();
for _ in 0..count {
let (tx, rx) = channel::<Reply>();
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 = create_reply();
let msg = Message::new(MsgType::Document);
reg.send(id, msg);
rx.recv_timeout(TIMEOUT).unwrap();
}
@ -184,7 +184,7 @@ mod clientregistries {
#[test]
fn prevent_duplicates() {
let mut reg = ClientRegistry::new();
let (tx, _rx) = channel::<Reply>();
let (tx, _rx) = channel::<Message>();
let existing = reg.add(tx);
let expected = Uuid::new_v4();
let ids = [existing.clone(), expected.clone()];
@ -208,12 +208,12 @@ impl ClientLink {
}
}
pub fn send(&mut self, req: Request) -> Receiver<Reply> {
pub fn send(&mut self, mut req: Message) -> Receiver<Message> {
let (tx, rx) = channel();
let mut msg: Message = req.into();
//let mut msg: Message = req.into();
let id = self.registry.add(tx);
msg.add_data("tx_id", id);
self.tx.send(msg).unwrap();
req.add_data("tx_id", id);
self.tx.send(req).unwrap();
rx
}
}
@ -232,7 +232,7 @@ mod clientlinks {
let mut registry = ClientRegistry::new();
let mut link = ClientLink::new(tx, registry.clone());
let req = Request::new(None);
let rx_client = link.send(req);
let rx_client = link.send(req.into());
let msg = rx.recv_timeout(TIMEOUT).unwrap();
match msg.get_msg_type() {
MsgType::ClientRequest => {}
@ -241,7 +241,7 @@ mod clientlinks {
match msg.get_data("tx_id") {
Some(result) => {
let id = result.to_uuid().unwrap();
registry.send(&id, create_reply());
registry.send(&id, msg.reply(MsgType::Document));
rx_client.recv().unwrap();
}
None => unreachable!("should have had a seender id"),
@ -319,11 +319,13 @@ impl Client {
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(),
);
self.registry.send(&tx_id, reply);
)s
*/
self.registry.send(&tx_id, initial_msg.reply(MsgType::Document));
}
}
@ -335,6 +337,7 @@ mod clients {
static TIMEOUT: Duration = Duration::from_millis(500);
/*
#[test]
fn start_client() {
let sess_id1 = Uuid::new_v4();
@ -348,7 +351,7 @@ mod clients {
);
let mut link = Client::start(queue.clone());
let req = get_root_with_session(&sess_id1);
let reply_rx = link.send(req);
let reply_rx = link.send(req.into());
let send1 = rx.recv_timeout(TIMEOUT).unwrap();
match send1.get_msg_type() {
MsgType::SessionValidate => {}
@ -376,7 +379,8 @@ mod clients {
document.add_data("doc", doc.clone());
queue.send(document).unwrap();
let reply = reply_rx.recv_timeout(TIMEOUT).unwrap();
assert_eq!(reply.get_session(), sess_id2);
assert_eq!(reply.get_content(), doc);
assert_eq!(reply.get_data("sess_id").unwrap().to_uuid().unwrap(), sess_id2);
assert_eq!(reply.get_data("doc").unwrap().to_string(), doc);
}
*/
}

21
src/lib.rs
View File

@ -10,8 +10,9 @@ use client::{Client, ClientLink, Reply, Request};
use clock::Clock;
use document::Document;
use field::Field;
use queue::Queue;
use queue::{Message, MsgType, Queue};
use session::Session;
use uuid::Uuid;
#[derive(Clone)]
pub struct MoreThanText {
@ -29,7 +30,21 @@ impl MoreThanText {
}
}
pub fn request<F>(&mut self, session: Option<F>) -> Reply
pub fn validate_session<F>(&mut self, session: Option<F>) -> Uuid
where
F: Into<Field>,
{
let mut msg = Message::new(MsgType::SessionValidate);
match session {
Some(id) => msg.add_data("sess_id", id.into()),
None => {},
}
let rx = self.client_link.send(msg);
let reply = rx.recv().unwrap();
reply.get_data("sess_id").unwrap().to_uuid().unwrap()
}
pub fn request<F>(&mut self, session: Option<F>) -> Message
where
F: Into<Field>,
{
@ -38,7 +53,7 @@ impl MoreThanText {
None => None,
};
let req = Request::new(sess);
let rx = self.client_link.send(req);
let rx = self.client_link.send(req.into());
rx.recv().unwrap()
}
}

38
src/main.rs
View File

@ -1,8 +1,19 @@
use axum::{extract::State, response::IntoResponse, routing::get, Router};
use axum::{
async_trait,
extract::{Extension, FromRequestParts, State},
http::request::Parts,
response::IntoResponse,
routing::get,
RequestPartsExt,
Router,
};
use axum_extra::extract::cookie::{Cookie, CookieJar};
use clap::Parser;
use morethantext::MoreThanText;
use std::convert::Infallible;
use tokio::{spawn, sync::mpsc::channel};
use tower_cookies::{CookieManagerLayer, Cookies};
use uuid::Uuid;
const LOCALHOST: &str = "127.0.0.1";
const SESSION_KEY: &str = "sessionid";
@ -33,11 +44,30 @@ async fn main() {
.unwrap();
}
struct SessionID;
#[async_trait]
impl<S> FromRequestParts<S> for SessionID
where
S: Send + Sync,
{
type Rejection = Infallible;
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
let cookies = parts.extract::<Extension<Cookies>>();
Ok(Self {})
}
}
async fn create_app(state: MoreThanText) -> Router {
Router::new().route("/", get(mtt_conn)).with_state(state)
}
async fn mtt_conn(jar: CookieJar, state: State<MoreThanText>) -> impl IntoResponse {
async fn mtt_conn(
jar: CookieJar,
sess_id: SessionID,
state: State<MoreThanText>,
) -> impl IntoResponse {
let sid = match jar.get(SESSION_KEY) {
Some(cookie) => Some(cookie.value().to_string()),
None => None,
@ -48,9 +78,9 @@ async fn mtt_conn(jar: CookieJar, state: State<MoreThanText>) -> impl IntoRespon
tx.send(state.clone().request(sess_info)).await.unwrap();
});
let reply = rx.recv().await.unwrap();
let cookie = Cookie::build((SESSION_KEY, reply.get_session().to_string()));
let cookie = Cookie::build((SESSION_KEY, reply.get_data("sess_id").unwrap().to_string()));
let cookies = jar.add(cookie);
(cookies, reply.get_content())
(cookies, reply.get_data("dov").unwrap().to_string())
}
#[cfg(test)]