remove event_register_handler from tcp stack

6 files changed, 20 insertions, 35 deletions

diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs
index 5d3f332..4e1249e 100644
--- a/mumctl/src/main.rs
+++ b/mumctl/src/main.rs
@@ -373,7 +373,7 @@ fn match_opt() -> Result<(), Error> {
             send_command(MumCommand::DeafenSelf(Some(false)))??;
         }
         Command::Messages => {
-            match send_command(MumCommand::PastMessages)?? {
+            match send_command(MumCommand::PastMessages { block: false })?? {
                 Some(CommandResponse::PastMessages { messages }) => {
                     for (msg, sender) in messages {
                         println!("{}: {}", sender, msg);
diff --git a/mumd/src/client.rs b/mumd/src/client.rs
index 9c2c2a0..3c491da 100644
--- a/mumd/src/client.rs
+++ b/mumd/src/client.rs
@@ -1,4 +1,4 @@
-use crate::command;
+use crate::{command, network::tcp::TcpEventQueue};
 use crate::error::ClientError;
 use crate::network::{tcp, udp, ConnectionInfo};
 use crate::state::State;
@@ -24,8 +24,7 @@ pub async fn handle(
         mpsc::unbounded_channel::<ControlPacket<Serverbound>>();
     let (ping_request_sender, ping_request_receiver) =
         mpsc::unbounded_channel();
-    let (response_sender, response_receiver) =
-        mpsc::unbounded_channel();
+    let event_queue = TcpEventQueue::new();
 
     let state = Arc::new(RwLock::new(state));
 
@@ -36,7 +35,7 @@ pub async fn handle(
             crypt_state_sender,
             packet_sender.clone(),
             packet_receiver,
-            response_receiver,
+            event_queue.clone(),
         ).fuse() => r.map_err(|e| ClientError::TcpError(e)),
         _ = udp::handle(
             Arc::clone(&state),
@@ -46,7 +45,7 @@ pub async fn handle(
         _ = command::handle(
             state,
             command_receiver,
-            response_sender,
+            event_queue,
             ping_request_sender,
             packet_sender,
             connection_info_sender,
diff --git a/mumd/src/command.rs b/mumd/src/command.rs
index 1337dce..a62ddbd 100644
--- a/mumd/src/command.rs
+++ b/mumd/src/command.rs
@@ -1,8 +1,4 @@
-use crate::network::{
-    ConnectionInfo,
-    tcp::{TcpEvent, TcpEventCallback},
-    udp::PingRequest
-};
+use crate::network::{ConnectionInfo, tcp::TcpEventQueue, udp::PingRequest};
 use crate::state::{ExecutionContext, State};
 
 use log::*;
@@ -17,7 +13,7 @@ pub async fn handle(
         Command,
         oneshot::Sender<mumlib::error::Result<Option<CommandResponse>>>,
     )>,
-    tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>,
+    tcp_event_queue: TcpEventQueue,
     ping_request_sender: mpsc::UnboundedSender<PingRequest>,
     mut packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
     mut connection_info_sender: watch::Sender<Option<ConnectionInfo>>,
@@ -33,14 +29,14 @@ pub async fn handle(
             ExecutionContext::TcpEvent(event, generator) => {
                 let (tx, rx) = oneshot::channel();
                 //TODO handle this error
-                let _ = tcp_event_register_sender.send((
-                    event,
+                tcp_event_queue.register_callback(
+                    event, 
                     Box::new(move |e| {
                         let response = generator(e);
                         response_sender.send(response).unwrap();
                         tx.send(()).unwrap();
                     }),
-                ));
+                );
 
                 rx.await.unwrap();
             }
diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs
index 8f34cd8..b6e939a 100644
--- a/mumd/src/network/tcp.rs
+++ b/mumd/src/network/tcp.rs
@@ -57,14 +57,14 @@ impl<'a> From<&TcpEventData<'a>> for TcpEvent {
 }
 
 #[derive(Clone)]
-struct TcpEventQueue {
+pub struct TcpEventQueue {
     callbacks: Arc<RwLock<HashMap<TcpEvent, Vec<TcpEventCallback>>>>,
     subscribers: Arc<RwLock<HashMap<TcpEvent, Vec<TcpEventSubscriber>>>>,
 }
 
 impl TcpEventQueue {
     /// Creates a new `TcpEventQueue`.
-    fn new() -> Self {
+    pub fn new() -> Self {
         Self {
             callbacks: Arc::new(RwLock::new(HashMap::new())),
             subscribers: Arc::new(RwLock::new(HashMap::new())),
@@ -72,18 +72,18 @@ impl TcpEventQueue {
     }
 
     /// Registers a new callback to be triggered when an event is fired.
-    fn register_callback(&self, at: TcpEvent, callback: TcpEventCallback) {
+    pub fn register_callback(&self, at: TcpEvent, callback: TcpEventCallback) {
         self.callbacks.write().unwrap().entry(at).or_default().push(callback);
     }
 
     /// Registers a new callback to be triggered when an event is fired.
-    fn register_subscriber(&self, at: TcpEvent, callback: TcpEventSubscriber) {
+    pub fn register_subscriber(&self, at: TcpEvent, callback: TcpEventSubscriber) {
         self.subscribers.write().unwrap().entry(at).or_default().push(callback);
     }
 
     /// Fires all callbacks related to a specific TCP event and removes them from the event queue.
     /// Also calls all event subscribers, but keeps them in the queue
-    fn resolve<'a>(&self, data: TcpEventData<'a>) {
+    pub fn resolve<'a>(&self, data: TcpEventData<'a>) {
         if let Some(vec) = self.callbacks.write().unwrap().get_mut(&TcpEvent::from(&data)) {
             let old = std::mem::take(vec);
             for handler in old {
@@ -107,7 +107,7 @@ pub async fn handle(
     crypt_state_sender: mpsc::Sender<ClientCryptState>,
     packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
     mut packet_receiver: mpsc::UnboundedReceiver<ControlPacket<Serverbound>>,
-    mut tcp_event_register_receiver: mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>,
+    event_queue: TcpEventQueue,
 ) -> Result<(), TcpError> {
     loop {
         let connection_info = 'data: loop {
@@ -137,7 +137,6 @@ pub async fn handle(
             (state_lock.phase_receiver(),
                 state_lock.audio_input().receiver())
         };
-        let event_queue = TcpEventQueue::new();
 
         info!("Logging in...");
 
@@ -160,7 +159,6 @@ pub async fn handle(
                         phase_watcher_inner,
                     ).fuse() => r,
                     r = send_packets(sink, &mut packet_receiver).fuse() => r,
-                    _ = register_events(&mut tcp_event_register_receiver, event_queue.clone()).fuse() => Ok(()),
                 }
             },
             phase_watcher,
@@ -403,13 +401,3 @@ async fn listen(
     }
     Ok(())
 }
-
-async fn register_events(
-    tcp_event_register_receiver: &mut mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>,
-    event_queue: TcpEventQueue,
-) {
-    loop {
-        let (event, handler) = tcp_event_register_receiver.recv().await.unwrap();
-        event_queue.register_callback(event, handler);
-    }
-}
diff --git a/mumd/src/state.rs b/mumd/src/state.rs
index bffc082..91c6ee7 100644
--- a/mumd/src/state.rs
+++ b/mumd/src/state.rs
@@ -393,7 +393,7 @@ impl State {
                 self.audio_output.set_user_volume(user_id, volume);
                 now!(Ok(None))
             }
-            Command::PastMessages => {
+            Command::PastMessages { block } => {
                 let server = match self.server.as_ref() {
                     Some(s) => s,
                     None => return now!(Err(Error::Disconnected)),
diff --git a/mumlib/src/command.rs b/mumlib/src/command.rs
index 5155aaa..4e8775f 100644
--- a/mumlib/src/command.rs
+++ b/mumlib/src/command.rs
@@ -29,7 +29,9 @@ pub enum Command {
     },
     Status,
     UserVolumeSet(String, f32),
-    PastMessages,
+    PastMessages {
+        block: bool,
+    },
     SendMessage {
         message: String,
         targets: Vec<MessageTarget>,