Merge branch 'network-rework' into 'main'

Add tcp event system

Closes #26

See merge request gustav/mum!17

4 files changed, 277 insertions, 215 deletions

diff --git a/mumd/src/command.rs b/mumd/src/command.rs
index a035a26..075bfaf 100644
--- a/mumd/src/command.rs
+++ b/mumd/src/command.rs
@@ -1,10 +1,11 @@
-use crate::state::{State, StatePhase};
+use crate::state::State;
 
 use ipc_channel::ipc::IpcSender;
 use log::*;
 use mumlib::command::{Command, CommandResponse};
 use std::sync::{Arc, Mutex};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
+use crate::network::tcp::{TcpEvent, TcpEventCallback};
 
 pub async fn handle(
     state: Arc<Mutex<State>>,
@@ -12,23 +13,26 @@ pub async fn handle(
         Command,
         IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
     )>,
+    tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>,
 ) {
     debug!("Begin listening for commands");
-    while let Some(command) = command_receiver.recv().await {
-        debug!("Received command {:?}", command.0);
+    while let Some((command, response_sender)) = command_receiver.recv().await {
+        debug!("Received command {:?}", command);
         let mut state = state.lock().unwrap();
-        let (wait_for_connected, command_response) = state.handle_command(command.0).await;
-        if wait_for_connected {
-            let mut watcher = state.phase_receiver();
-            drop(state);
-            while !matches!(watcher.recv().await.unwrap(), StatePhase::Connected) {}
+        let (event, generator) = state.handle_command(command).await;
+        drop(state);
+        if let Some(event) = event {
+            let (tx, rx) = oneshot::channel();
+            //TODO handle this error
+            let _ = tcp_event_register_sender.send((event, Box::new(move |e| {
+                let response = generator(Some(e));
+                response_sender.send(response).unwrap();
+                tx.send(()).unwrap();
+            })));
+
+            rx.await.unwrap();
+        } else {
+            response_sender.send(generator(None)).unwrap();
         }
-        command.1.send(command_response).unwrap();
     }
-    //TODO err if not connected
-    //while let Some(command) = command_receiver.recv().await {
-    //    debug!("Parsing command {:?}", command);
-    //}
-
-    //debug!("Finished handling commands");
 }
diff --git a/mumd/src/main.rs b/mumd/src/main.rs
index 75726f8..e88eede 100644
--- a/mumd/src/main.rs
+++ b/mumd/src/main.rs
@@ -33,6 +33,7 @@ async fn main() {
     )>();
     let (connection_info_sender, connection_info_receiver) =
         watch::channel::<Option<ConnectionInfo>>(None);
+    let (response_sender, response_receiver) = mpsc::unbounded_channel();
 
     let state = State::new(packet_sender, connection_info_sender);
     let state = Arc::new(Mutex::new(state));
@@ -43,13 +44,14 @@ async fn main() {
             connection_info_receiver.clone(),
             crypt_state_sender,
             packet_receiver,
+            response_receiver,
         ),
         network::udp::handle(
             Arc::clone(&state),
             connection_info_receiver.clone(),
             crypt_state_receiver,
         ),
-        command::handle(state, command_receiver,),
+        command::handle(state, command_receiver, response_sender),
         spawn_blocking(move || {
             // IpcSender is blocking
             receive_oneshot_commands(command_sender);
diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs
index 6471771..c2cb234 100644
--- a/mumd/src/network/tcp.rs
+++ b/mumd/src/network/tcp.rs
@@ -15,6 +15,10 @@ use tokio::sync::{mpsc, oneshot, watch};
 use tokio::time::{self, Duration};
 use tokio_tls::{TlsConnector, TlsStream};
 use tokio_util::codec::{Decoder, Framed};
+use std::collections::HashMap;
+use std::future::Future;
+use std::rc::Rc;
+use std::cell::RefCell;
 
 type TcpSender = SplitSink<
     Framed<TlsStream<TcpStream>, ControlCodec<Serverbound, Clientbound>>,
@@ -23,11 +27,25 @@ type TcpSender = SplitSink<
 type TcpReceiver =
     SplitStream<Framed<TlsStream<TcpStream>, ControlCodec<Serverbound, Clientbound>>>;
 
+pub(crate) type TcpEventCallback = Box<dyn FnOnce(&TcpEventData)>;
+
+#[derive(Debug, Clone, Hash, Eq, PartialEq)]
+pub enum TcpEvent {
+    Connected, //fires when the client has connected to a server
+    Disconnected, //fires when the client has disconnected from a server
+}
+
+pub enum TcpEventData<'a> {
+    Connected(&'a msgs::ServerSync),
+    Disconnected,
+}
+
 pub async fn handle(
     state: Arc<Mutex<State>>,
     mut connection_info_receiver: watch::Receiver<Option<ConnectionInfo>>,
     crypt_state_sender: mpsc::Sender<ClientCryptState>,
     mut packet_receiver: mpsc::UnboundedReceiver<ControlPacket<Serverbound>>,
+    mut tcp_event_register_receiver: mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>,
 ) {
     loop {
         let connection_info = loop {
@@ -54,6 +72,7 @@ pub async fn handle(
         let phase_watcher = state_lock.phase_receiver();
         let packet_sender = state_lock.packet_sender();
         drop(state_lock);
+        let event_queue = Arc::new(Mutex::new(HashMap::new()));
 
         info!("Logging in...");
 
@@ -63,9 +82,11 @@ pub async fn handle(
                 Arc::clone(&state),
                 stream,
                 crypt_state_sender.clone(),
-                phase_watcher.clone()
+                Arc::clone(&event_queue),
+                phase_watcher.clone(),
             ),
-            send_packets(sink, &mut packet_receiver, phase_watcher),
+            send_packets(sink, &mut packet_receiver, phase_watcher.clone()),
+            register_events(&mut tcp_event_register_receiver, event_queue, phase_watcher),
         );
 
         debug!("Fully disconnected TCP stream, waiting for new connection info");
@@ -108,103 +129,207 @@ async fn authenticate(sink: &mut TcpSender, username: String) {
 async fn send_pings(
     packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
     delay_seconds: u64,
-    mut phase_watcher: watch::Receiver<StatePhase>,
+    phase_watcher: watch::Receiver<StatePhase>,
 ) {
-    let (tx, rx) = oneshot::channel();
-    let phase_transition_block = async {
-        while !matches!(
-            phase_watcher.recv().await.unwrap(),
-            StatePhase::Disconnected
-        ) {}
-        tx.send(true).unwrap();
-    };
-
-    let mut interval = time::interval(Duration::from_secs(delay_seconds));
-    let main_block = async {
-        let rx = rx.fuse();
-        pin_mut!(rx);
-        loop {
-            let interval_waiter = interval.tick().fuse();
-            pin_mut!(interval_waiter);
-            let exitor = select! {
-                data = interval_waiter => Some(data),
-                _ = rx => None
-            };
-
-            match exitor {
-                Some(_) => {
-                    trace!("Sending ping");
-                    let msg = msgs::Ping::new();
-                    packet_sender.send(msg.into()).unwrap();
-                }
-                None => break,
-            }
-        }
-    };
+    let interval = Rc::new(RefCell::new(time::interval(Duration::from_secs(delay_seconds))));
+    let packet_sender = Rc::new(RefCell::new(packet_sender));
 
-    join!(main_block, phase_transition_block);
+    run_until_disconnection(
+        || async {
+            Some(interval.borrow_mut().tick().await)
+        },
+        |_| async {
+            trace!("Sending ping");
+            let msg = msgs::Ping::new();
+            packet_sender.borrow_mut().send(msg.into()).unwrap();
+        },
+        || async {},
+        phase_watcher,
+    ).await;
 
     debug!("Ping sender process killed");
 }
 
 async fn send_packets(
-    mut sink: TcpSender,
+    sink: TcpSender,
     packet_receiver: &mut mpsc::UnboundedReceiver<ControlPacket<Serverbound>>,
-    mut phase_watcher: watch::Receiver<StatePhase>,
+    phase_watcher: watch::Receiver<StatePhase>,
 ) {
-    let (tx, rx) = oneshot::channel();
-    let phase_transition_block = async {
-        while !matches!(
-            phase_watcher.recv().await.unwrap(),
-            StatePhase::Disconnected
-        ) {}
-        tx.send(true).unwrap();
-    };
-
-    let main_block = async {
-        let rx = rx.fuse();
-        pin_mut!(rx);
-        loop {
-            let packet_recv = packet_receiver.recv().fuse();
-            pin_mut!(packet_recv);
-            let exitor = select! {
-                data = packet_recv => Some(data),
-                _ = rx => None
-            };
-            match exitor {
-                None => {
-                    break;
-                }
-                Some(None) => {
-                    warn!("Channel closed before disconnect command");
-                    break;
-                }
-                Some(Some(packet)) => {
-                    sink.send(packet).await.unwrap();
-                }
-            }
-        }
+    let sink = Rc::new(RefCell::new(sink));
+    let packet_receiver = Rc::new(RefCell::new(packet_receiver));
+    run_until_disconnection(
+        || async {
+            packet_receiver.borrow_mut().recv().await
+        },
+        |packet| async {
+            sink.borrow_mut().send(packet).await.unwrap();
+        },
+        || async {
+            //clears queue of remaining packets
+            while packet_receiver.borrow_mut().try_recv().is_ok() {}
 
-        //clears queue of remaining packets
-        while packet_receiver.try_recv().is_ok() {}
-
-        sink.close().await.unwrap();
-    };
-
-    join!(main_block, phase_transition_block);
+            sink.borrow_mut().close().await.unwrap();
+        },
+        phase_watcher,
+    ).await;
 
     debug!("TCP packet sender killed");
 }
 
 async fn listen(
     state: Arc<Mutex<State>>,
-    mut stream: TcpReceiver,
+    stream: TcpReceiver,
     crypt_state_sender: mpsc::Sender<ClientCryptState>,
-    mut phase_watcher: watch::Receiver<StatePhase>,
+    event_queue: Arc<Mutex<HashMap<TcpEvent, Vec<TcpEventCallback>>>>,
+    phase_watcher: watch::Receiver<StatePhase>,
+) {
+    let crypt_state = Rc::new(RefCell::new(None));
+    let crypt_state_sender = Rc::new(RefCell::new(Some(crypt_state_sender)));
+
+    let stream = Rc::new(RefCell::new(stream));
+    run_until_disconnection(
+        || async {
+            stream.borrow_mut().next().await
+        },
+        |packet| async {
+            match packet.unwrap() {
+                ControlPacket::TextMessage(msg) => {
+                    info!(
+                        "Got message from user with session ID {}: {}",
+                        msg.get_actor(),
+                        msg.get_message()
+                    );
+                }
+                ControlPacket::CryptSetup(msg) => {
+                    debug!("Crypt setup");
+                    // Wait until we're fully connected before initiating UDP voice
+                    *crypt_state.borrow_mut() = Some(ClientCryptState::new_from(
+                        msg.get_key()
+                            .try_into()
+                            .expect("Server sent private key with incorrect size"),
+                        msg.get_client_nonce()
+                            .try_into()
+                            .expect("Server sent client_nonce with incorrect size"),
+                        msg.get_server_nonce()
+                            .try_into()
+                            .expect("Server sent server_nonce with incorrect size"),
+                    ));
+                }
+                ControlPacket::ServerSync(msg) => {
+                    info!("Logged in");
+                    if let Some(mut sender) = crypt_state_sender.borrow_mut().take() {
+                        let _ = sender
+                            .send(
+                                crypt_state
+                                    .borrow_mut()
+                                    .take()
+                                    .expect("Server didn't send us any CryptSetup packet!"),
+                            )
+                            .await;
+                    }
+                    if let Some(vec) = event_queue.lock().unwrap().get_mut(&TcpEvent::Connected) {
+                        let old = std::mem::take(vec);
+                        for handler in old {
+                            handler(&TcpEventData::Connected(&msg));
+                        }
+                    }
+                    let mut state = state.lock().unwrap();
+                    let server = state.server_mut().unwrap();
+                    server.parse_server_sync(*msg);
+                    match &server.welcome_text {
+                        Some(s) => info!("Welcome: {}", s),
+                        None => info!("No welcome received"),
+                    }
+                    for channel in server.channels().values() {
+                        info!("Found channel {}", channel.name());
+                    }
+                    state.initialized();
+                }
+                ControlPacket::Reject(msg) => {
+                    warn!("Login rejected: {:?}", msg);
+                }
+                ControlPacket::UserState(msg) => {
+                    let mut state = state.lock().unwrap();
+                    let session = msg.get_session();
+                    if *state.phase_receiver().borrow() == StatePhase::Connecting {
+                        state.audio_mut().add_client(msg.get_session());
+                        state.parse_user_state(*msg);
+                    } else {
+                        state.parse_user_state(*msg);
+                    }
+                    let server = state.server_mut().unwrap();
+                    let user = server.users().get(&session).unwrap();
+                    info!("User {} connected to {}", user.name(), user.channel());
+                }
+                ControlPacket::UserRemove(msg) => {
+                    info!("User {} left", msg.get_session());
+                    state
+                        .lock()
+                        .unwrap()
+                        .audio_mut()
+                        .remove_client(msg.get_session());
+                }
+                ControlPacket::ChannelState(msg) => {
+                    debug!("Channel state received");
+                    state
+                        .lock()
+                        .unwrap()
+                        .server_mut()
+                        .unwrap()
+                        .parse_channel_state(*msg); //TODO parse initial if initial
+                }
+                ControlPacket::ChannelRemove(msg) => {
+                    state
+                        .lock()
+                        .unwrap()
+                        .server_mut()
+                        .unwrap()
+                        .parse_channel_remove(*msg);
+                }
+                _ => {}
+            }
+        },
+        || async {
+            if let Some(vec) = event_queue.lock().unwrap().get_mut(&TcpEvent::Disconnected) {
+                let old = std::mem::take(vec);
+                for handler in old {
+                    handler(&TcpEventData::Disconnected);
+                }
+            }
+        },
+        phase_watcher,
+    ).await;
+
+    debug!("Killing TCP listener block");
+}
+
+async fn register_events(
+    tcp_event_register_receiver: &mut mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>,
+    event_data: Arc<Mutex<HashMap<TcpEvent, Vec<TcpEventCallback>>>>,
+    phase_watcher: watch::Receiver<StatePhase>,
 ) {
-    let mut crypt_state = None;
-    let mut crypt_state_sender = Some(crypt_state_sender);
+    let tcp_event_register_receiver = Rc::new(RefCell::new(tcp_event_register_receiver));
+    run_until_disconnection(
+        || async {
+            tcp_event_register_receiver.borrow_mut().recv().await
+        },
+        |(event, handler)| async { event_data.lock().unwrap().entry(event).or_default().push(handler); },
+        || async {},
+        phase_watcher,
+    ).await;
+}
 
+async fn run_until_disconnection<T, F, G, H>(
+    mut generator: impl FnMut() -> F,
+    mut handler: impl FnMut(T) -> G,
+    mut shutdown: impl FnMut() -> H,
+    mut phase_watcher: watch::Receiver<StatePhase>,
+)
+    where
+        F: Future<Output = Option<T>>,
+        G: Future<Output = ()>,
+        H: Future<Output = ()>,
+{
     let (tx, rx) = oneshot::channel();
     let phase_transition_block = async {
         while !matches!(
@@ -214,11 +339,11 @@ async fn listen(
         tx.send(true).unwrap();
     };
 
-    let listener_block = async {
+    let main_block = async {
         let rx = rx.fuse();
         pin_mut!(rx);
         loop {
-            let packet_recv = stream.next().fuse();
+            let packet_recv = generator().fuse();
             pin_mut!(packet_recv);
             let exitor = select! {
                 data = packet_recv => Some(data),
@@ -229,101 +354,17 @@ async fn listen(
                     break;
                 }
                 Some(None) => {
-                    warn!("Channel closed before disconnect command");
+                    //warn!("Channel closed before disconnect command"); //TODO make me informative
                     break;
                 }
-                Some(Some(packet)) => {
-                    match packet.unwrap() {
-                        ControlPacket::TextMessage(msg) => {
-                            info!(
-                                "Got message from user with session ID {}: {}",
-                                msg.get_actor(),
-                                msg.get_message()
-                            );
-                        }
-                        ControlPacket::CryptSetup(msg) => {
-                            debug!("Crypt setup");
-                            // Wait until we're fully connected before initiating UDP voice
-                            crypt_state = Some(ClientCryptState::new_from(
-                                msg.get_key()
-                                    .try_into()
-                                    .expect("Server sent private key with incorrect size"),
-                                msg.get_client_nonce()
-                                    .try_into()
-                                    .expect("Server sent client_nonce with incorrect size"),
-                                msg.get_server_nonce()
-                                    .try_into()
-                                    .expect("Server sent server_nonce with incorrect size"),
-                            ));
-                        }
-                        ControlPacket::ServerSync(msg) => {
-                            info!("Logged in");
-                            if let Some(mut sender) = crypt_state_sender.take() {
-                                let _ = sender
-                                    .send(
-                                        crypt_state
-                                            .take()
-                                            .expect("Server didn't send us any CryptSetup packet!"),
-                                    )
-                                    .await;
-                            }
-                            let mut state = state.lock().unwrap();
-                            let server = state.server_mut().unwrap();
-                            server.parse_server_sync(*msg);
-                            match &server.welcome_text {
-                                Some(s) => info!("Welcome: {}", s),
-                                None => info!("No welcome received"),
-                            }
-                            for channel in server.channels().values() {
-                                info!("Found channel {}", channel.name());
-                            }
-                            state.initialized();
-                        }
-                        ControlPacket::Reject(msg) => {
-                            warn!("Login rejected: {:?}", msg);
-                        }
-                        ControlPacket::UserState(msg) => {
-                            let mut state = state.lock().unwrap();
-                            let session = msg.get_session();
-
-                            let user_state_diff = state.parse_user_state(*msg);
-                            //TODO do something with user state diff
-                            debug!("user state diff: {:#?}", &user_state_diff);
-
-                            let server = state.server_mut().unwrap();
-                            let user = server.users().get(&session).unwrap();
-                            info!("User {} connected to {}", user.name(), user.channel());
-                        }
-                        ControlPacket::UserRemove(msg) => {
-                            state.lock().unwrap().remove_client(*msg);
-                        }
-                        ControlPacket::ChannelState(msg) => {
-                            debug!("Channel state received");
-                            state
-                                .lock()
-                                .unwrap()
-                                .server_mut()
-                                .unwrap()
-                                .parse_channel_state(*msg); //TODO parse initial if initial
-                        }
-                        ControlPacket::ChannelRemove(msg) => {
-                            state
-                                .lock()
-                                .unwrap()
-                                .server_mut()
-                                .unwrap()
-                                .parse_channel_remove(*msg);
-                        }
-                        _ => {}
-                    }
+                Some(Some(data)) => {
+                    handler(data).await;
                 }
             }
         }
 
-        //TODO? clean up stream
+        shutdown().await;
     };
 
-    join!(phase_transition_block, listener_block);
-
-    debug!("Killing TCP listener block");
-}
+    join!(main_block, phase_transition_block);
+}
\ No newline at end of file
diff --git a/mumd/src/state.rs b/mumd/src/state.rs
index e9db616..1a02068 100644
--- a/mumd/src/state.rs
+++ b/mumd/src/state.rs
@@ -14,6 +14,7 @@ use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::net::ToSocketAddrs;
 use tokio::sync::{mpsc, watch};
+use crate::network::tcp::{TcpEvent, TcpEventData};
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum StatePhase {
@@ -55,11 +56,11 @@ impl State {
     pub async fn handle_command(
         &mut self,
         command: Command,
-    ) -> (bool, mumlib::error::Result<Option<CommandResponse>>) {
+    ) -> (Option<TcpEvent>, Box<dyn FnOnce(Option<&TcpEventData>) -> mumlib::error::Result<Option<CommandResponse>>>) {
         match command {
             Command::ChannelJoin { channel_identifier } => {
                 if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
-                    return (false, Err(Error::DisconnectedError));
+                    return (None, Box::new(|_| Err(Error::DisconnectedError)));
                 }
 
                 let channels = self.server()
@@ -77,33 +78,34 @@ impl State {
                             .filter(|e| e.1.ends_with(&channel_identifier.to_lowercase()))
                             .collect::<Vec<_>>();
                         match soft_matches.len() {
-                            0 => return (false, Err(Error::ChannelIdentifierError(channel_identifier, ChannelIdentifierError::Invalid))),
+                            0 => return (None, Box::new(|_| Err(Error::ChannelIdentifierError(channel_identifier, ChannelIdentifierError::Invalid)))),
                             1 => *soft_matches.get(0).unwrap().0,
-                            _ => return (false, Err(Error::ChannelIdentifierError(channel_identifier, ChannelIdentifierError::Invalid))),
+                            _ => return (None, Box::new(|_| Err(Error::ChannelIdentifierError(channel_identifier, ChannelIdentifierError::Invalid)))),
                         }
                     },
                     1 => *matches.get(0).unwrap().0,
-                    _ => return (false, Err(Error::ChannelIdentifierError(channel_identifier, ChannelIdentifierError::Ambiguous))),
+                    _ => return (None, Box::new(|_| Err(Error::ChannelIdentifierError(channel_identifier, ChannelIdentifierError::Ambiguous)))),
                 };
 
                 let mut msg = msgs::UserState::new();
                 msg.set_session(self.server.as_ref().unwrap().session_id.unwrap());
                 msg.set_channel_id(id);
                 self.packet_sender.send(msg.into()).unwrap();
-                (false, Ok(None))
+                (None, Box::new(|_| Ok(None)))
             }
             Command::ChannelList => {
                 if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
-                    return (false, Err(Error::DisconnectedError));
+                    return (None, Box::new(|_| Err(Error::DisconnectedError)));
                 }
+                let list = into_channel(
+                    self.server.as_ref().unwrap().channels(),
+                    self.server.as_ref().unwrap().users(),
+                );
                 (
-                    false,
-                    Ok(Some(CommandResponse::ChannelList {
-                        channels: into_channel(
-                            self.server.as_ref().unwrap().channels(),
-                            self.server.as_ref().unwrap().users(),
-                        ),
-                    })),
+                    None,
+                    Box::new(move |_| Ok(Some(CommandResponse::ChannelList {
+                        channels: list,
+                    }))),
                 )
             }
             Command::ServerConnect {
@@ -113,7 +115,7 @@ impl State {
                 accept_invalid_cert,
             } => {
                 if !matches!(*self.phase_receiver().borrow(), StatePhase::Disconnected) {
-                    return (false, Err(Error::AlreadyConnectedError));
+                    return (None, Box::new(|_| Err(Error::AlreadyConnectedError)));
                 }
                 let mut server = Server::new();
                 server.username = Some(username);
@@ -131,7 +133,7 @@ impl State {
                     Ok(Some(v)) => v,
                     _ => {
                         warn!("Error parsing server addr");
-                        return (false, Err(Error::InvalidServerAddrError(host, port)));
+                        return (None, Box::new(move |_| Err(Error::InvalidServerAddrError(host, port))));
                     }
                 };
                 self.connection_info_sender
@@ -141,22 +143,35 @@ impl State {
                         accept_invalid_cert,
                     )))
                     .unwrap();
-                (true, Ok(None))
+                (Some(TcpEvent::Connected), Box::new(|e| { //runs the closure when the client is connected
+                    if let Some(TcpEventData::Connected(msg)) = e {
+                        Ok(Some(CommandResponse::ServerConnect {
+                            welcome_message: if msg.has_welcome_text() {
+                                Some(msg.get_welcome_text().to_string())
+                            } else {
+                                None
+                            }
+                        }))
+                    } else {
+                        unreachable!("callback should be provided with a TcpEventData::Connected");
+                    }
+                }))
             }
             Command::Status => {
                 if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
-                    return (false, Err(Error::DisconnectedError));
+                    return (None, Box::new(|_| Err(Error::DisconnectedError)));
                 }
+                let state = self.server.as_ref().unwrap().into();
                 (
-                    false,
-                    Ok(Some(CommandResponse::Status {
-                        server_state: self.server.as_ref().unwrap().into(), //guaranteed not to panic because if we are connected, server is guaranteed to be Some
-                    })),
+                    None,
+                    Box::new(move |_| Ok(Some(CommandResponse::Status {
+                        server_state: state, //guaranteed not to panic because if we are connected, server is guaranteed to be Some
+                    }))),
                 )
             }
             Command::ServerDisconnect => {
                 if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
-                    return (false, Err(Error::DisconnectedError));
+                    return (None, Box::new(|_| Err(Error::DisconnectedError)));
                 }
 
                 self.server = None;
@@ -166,15 +181,15 @@ impl State {
                     .0
                     .broadcast(StatePhase::Disconnected)
                     .unwrap();
-                (false, Ok(None))
+                (None, Box::new(|_| Ok(None)))
             }
             Command::InputVolumeSet(volume) => {
                 self.audio.set_input_volume(volume);
-                (false, Ok(None))
+                (None, Box::new(|_| Ok(None)))
             }
             Command::ConfigReload => {
                 self.reload_config();
-                (false, Ok(None))
+                (None, Box::new(|_| Ok(None)))
             }
         }
     }