Attached queue to client.

This commit is contained in:
Jeff Baskin 2025-04-03 22:28:47 -04:00
parent dc9aab3bab
commit 1f9ea9b170
2 changed files with 22 additions and 11 deletions

View File

@ -194,21 +194,23 @@ mod clientlinks {
} }
pub struct Client { pub struct Client {
queue: Queue,
registry: ClientRegistry, registry: ClientRegistry,
rx: Receiver<Message>, rx: Receiver<Message>,
} }
impl Client { impl Client {
fn new(rx: Receiver<Message>) -> Self { fn new(rx: Receiver<Message>, queue: Queue) -> Self {
Self { Self {
queue: queue,
registry: ClientRegistry::new(), registry: ClientRegistry::new(),
rx: rx, rx: rx,
} }
} }
pub fn start(_queue: Queue) -> ClientLink { pub fn start(queue: Queue) -> ClientLink {
let (tx, rx) = channel(); let (tx, rx) = channel();
let mut client = Client::new(rx); let mut client = Client::new(rx, queue);
let link = ClientLink::new(tx, client.get_registry()); let link = ClientLink::new(tx, client.get_registry());
spawn(move || { spawn(move || {
client.listen(); client.listen();
@ -219,6 +221,7 @@ impl Client {
fn listen(&mut self) { fn listen(&mut self) {
loop { loop {
let msg = self.rx.recv().unwrap(); let msg = self.rx.recv().unwrap();
self.queue.send(Message::new(MsgType::SessionValidate));
let id = msg.get_data().get("tx_id").unwrap().to_uuid().unwrap(); let id = msg.get_data().get("tx_id").unwrap().to_uuid().unwrap();
let reply = Reply {}; let reply = Reply {};
self.registry.send(&id, reply); self.registry.send(&id, reply);
@ -234,12 +237,22 @@ impl Client {
mod clients { mod clients {
use super::*; use super::*;
use requests::create_request; use requests::create_request;
use std::time::Duration;
static TIMEOUT: Duration = Duration::from_millis(500);
#[test] #[test]
fn start_client() { fn start_client() {
let queue = Queue::new(); let (tx, rx) = channel();
let mut queue = Queue::new();
queue.add(tx);
let mut link = Client::start(queue.clone()); let mut link = Client::start(queue.clone());
let req = create_request(); let req = create_request();
link.send(req); link.send(req);
let sess = rx.recv_timeout(TIMEOUT).unwrap();
match sess.get_class() {
MsgType::SessionValidate => {},
_ => unreachable!("should request session validation"),
}
} }
} }

View File

@ -1,10 +1,7 @@
use crate::{client::Request, field::Field}; use crate::{client::Request, field::Field};
use std::{ use std::{
collections::HashMap, collections::HashMap,
sync::{ sync::{mpsc::Sender, Arc, RwLock},
mpsc::Sender,
Arc, RwLock,
},
}; };
use uuid::Uuid; use uuid::Uuid;
@ -12,6 +9,7 @@ use uuid::Uuid;
pub enum MsgType { pub enum MsgType {
ClientRequest, ClientRequest,
NoOp, NoOp,
SessionValidate,
} }
#[derive(Clone)] #[derive(Clone)]
@ -137,12 +135,12 @@ impl Queue {
} }
} }
fn add(&self, tx: Sender<Message>) { pub fn add(&self, tx: Sender<Message>) {
let mut store = self.store.write().unwrap(); let mut store = self.store.write().unwrap();
store.push(tx); store.push(tx);
} }
fn send(&self, msg: Message) { pub fn send(&self, msg: Message) {
let store = self.store.read().unwrap(); let store = self.store.read().unwrap();
for sender in store.iter() { for sender in store.iter() {
sender.send(msg.clone()).unwrap(); sender.send(msg.clone()).unwrap();
@ -152,8 +150,8 @@ impl Queue {
#[cfg(test)] #[cfg(test)]
mod serviceredistries { mod serviceredistries {
use std::sync::mpsc::channel;
use super::*; use super::*;
use std::sync::mpsc::channel;
#[test] #[test]
fn create_queue() { fn create_queue() {