Merge branch 'main' into noise-gate

9 files changed, 104 insertions, 90 deletions

diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs
index bf3f7f4..83818d5 100644
--- a/mumd/src/audio.rs
+++ b/mumd/src/audio.rs
@@ -25,7 +25,7 @@ use tokio::sync::watch;
 use mumble_protocol::Serverbound;
 
 //TODO? move to mumlib
-pub const EVENT_SOUNDS: &[(&'static [u8], NotificationEvents)] = &[
+pub const EVENT_SOUNDS: &[(&[u8], NotificationEvents)] = &[
     (include_bytes!("resources/connect.wav"), NotificationEvents::ServerConnect),
     (
         include_bytes!("resources/disconnect.wav"),
diff --git a/mumd/src/client.rs b/mumd/src/client.rs
new file mode 100644
index 0000000..3613061
--- /dev/null
+++ b/mumd/src/client.rs
@@ -0,0 +1,55 @@
+use crate::command;
+use crate::network::{tcp, udp, ConnectionInfo};
+use crate::state::State;
+
+use futures_util::join;
+use ipc_channel::ipc::IpcSender;
+use mumble_protocol::{Serverbound, control::ControlPacket, crypt::ClientCryptState};
+use mumlib::command::{Command, CommandResponse};
+use std::sync::{Arc, Mutex};
+use tokio::sync::{mpsc, watch};
+
+pub async fn handle(
+    command_receiver: mpsc::UnboundedReceiver<(
+        Command,
+        IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
+    )>,
+) {
+    let (connection_info_sender, connection_info_receiver) =
+        watch::channel::<Option<ConnectionInfo>>(None);
+    let (crypt_state_sender, crypt_state_receiver) =
+        mpsc::channel::<ClientCryptState>(1);
+    let (packet_sender, packet_receiver) =
+        mpsc::unbounded_channel::<ControlPacket<Serverbound>>();
+    let (ping_request_sender, ping_request_receiver) =
+        mpsc::unbounded_channel();
+    let (response_sender, response_receiver) =
+        mpsc::unbounded_channel();
+
+    let state = State::new();
+    let state = Arc::new(Mutex::new(state));
+    join!(
+        tcp::handle(
+            Arc::clone(&state),
+            connection_info_receiver.clone(),
+            crypt_state_sender,
+            packet_sender.clone(),
+            packet_receiver,
+            response_receiver,
+        ),
+        udp::handle(
+            Arc::clone(&state),
+            connection_info_receiver.clone(),
+            crypt_state_receiver,
+        ),
+        command::handle(
+            state,
+            command_receiver,
+            response_sender,
+            ping_request_sender,
+            packet_sender,
+            connection_info_sender,
+        ),
+        udp::handle_pings(ping_request_receiver),
+    );
+}
diff --git a/mumd/src/command.rs b/mumd/src/command.rs
index 330e3fc..e77b34b 100644
--- a/mumd/src/command.rs
+++ b/mumd/src/command.rs
@@ -1,13 +1,16 @@
+use crate::network::{
+    ConnectionInfo,
+    tcp::{TcpEvent, TcpEventCallback},
+    udp::PingRequest
+};
 use crate::state::{ExecutionContext, State};
 
-use crate::network::tcp::{TcpEvent, TcpEventCallback};
 use ipc_channel::ipc::IpcSender;
 use log::*;
-use mumble_protocol::ping::PongPacket;
+use mumble_protocol::{Serverbound, control::ControlPacket};
 use mumlib::command::{Command, CommandResponse};
-use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{mpsc, oneshot, watch};
 
 pub async fn handle(
     state: Arc<Mutex<State>>,
@@ -16,13 +19,15 @@ pub async fn handle(
         IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
     )>,
     tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>,
-    ping_request_sender: mpsc::UnboundedSender<(u64, SocketAddr, Box<dyn FnOnce(PongPacket)>)>,
+    ping_request_sender: mpsc::UnboundedSender<PingRequest>,
+    mut packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
+    mut connection_info_sender: watch::Sender<Option<ConnectionInfo>>,
 ) {
     debug!("Begin listening for commands");
     while let Some((command, response_sender)) = command_receiver.recv().await {
         debug!("Received command {:?}", command);
         let mut state = state.lock().unwrap();
-        let event = state.handle_command(command);
+        let event = state.handle_command(command, &mut packet_sender, &mut connection_info_sender);
         drop(state);
         match event {
             ExecutionContext::TcpEvent(event, generator) => {
diff --git a/mumd/src/main.rs b/mumd/src/main.rs
index db6d2ef..67481f9 100644
--- a/mumd/src/main.rs
+++ b/mumd/src/main.rs
@@ -1,23 +1,17 @@
 mod audio;
+mod client;
 mod command;
 mod network;
 mod notify;
 mod state;
 
-use crate::network::ConnectionInfo;
-use crate::state::State;
-
 use futures::join;
 use ipc_channel::ipc::{self, IpcOneShotServer, IpcSender};
 use log::*;
-use mumble_protocol::control::ControlPacket;
-use mumble_protocol::crypt::ClientCryptState;
-use mumble_protocol::voice::Serverbound;
 use mumlib::command::{Command, CommandResponse};
 use mumlib::setup_logger;
 use std::fs;
-use std::sync::{Arc, Mutex};
-use tokio::sync::{mpsc, watch};
+use tokio::sync::mpsc;
 use tokio::task::spawn_blocking;
 
 #[tokio::main]
@@ -45,46 +39,17 @@ async fn main() {
         }
     }
 
-    // Oneshot channel for setting UDP CryptState from control task
-    // For simplicity we don't deal with re-syncing, real applications would have to.
-    let (crypt_state_sender, crypt_state_receiver) = mpsc::channel::<ClientCryptState>(1); // crypt state should always be consumed before sending a new one
-    let (packet_sender, packet_receiver) = mpsc::unbounded_channel::<ControlPacket<Serverbound>>();
     let (command_sender, command_receiver) = mpsc::unbounded_channel::<(
         Command,
         IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
     )>();
-    let (connection_info_sender, connection_info_receiver) =
-        watch::channel::<Option<ConnectionInfo>>(None);
-    let (response_sender, response_receiver) = mpsc::unbounded_channel();
-    let (ping_request_sender, ping_request_receiver) = mpsc::unbounded_channel();
-
-    let state = State::new(packet_sender, connection_info_sender);
-    let state = Arc::new(Mutex::new(state));
 
-    let (_, _, _, e, _) = join!(
-        network::tcp::handle(
-            Arc::clone(&state),
-            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,
-            response_sender,
-            ping_request_sender,
-        ),
+    let (_, e) = join!(
+        client::handle(command_receiver),
         spawn_blocking(move || {
             // IpcSender is blocking
             receive_oneshot_commands(command_sender);
         }),
-        network::udp::handle_pings(ping_request_receiver),
     );
     e.unwrap();
 }
diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs
index 3c96ee1..47ea311 100644
--- a/mumd/src/network/tcp.rs
+++ b/mumd/src/network/tcp.rs
@@ -44,6 +44,7 @@ pub async fn handle(
     state: Arc<Mutex<State>>,
     mut connection_info_receiver: watch::Receiver<Option<ConnectionInfo>>,
     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)>,
 ) {
@@ -67,14 +68,13 @@ pub async fn handle(
         let state_lock = state.lock().unwrap();
         authenticate(&mut sink, state_lock.username().unwrap().to_string()).await;
         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...");
 
         join!(
-            send_pings(packet_sender, 10, phase_watcher.clone()),
+            send_pings(packet_sender.clone(), 10, phase_watcher.clone()),
             listen(
                 Arc::clone(&state),
                 stream,
diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs
index 4dde268..0c00029 100644
--- a/mumd/src/network/udp.rs
+++ b/mumd/src/network/udp.rs
@@ -1,10 +1,10 @@
 use crate::network::ConnectionInfo;
 use crate::state::{State, StatePhase};
-use log::*;
 
 use bytes::Bytes;
 use futures::{join, pin_mut, select, FutureExt, SinkExt, StreamExt, Stream};
 use futures_util::stream::{SplitSink, SplitStream};
+use log::*;
 use mumble_protocol::crypt::ClientCryptState;
 use mumble_protocol::ping::{PingPacket, PongPacket};
 use mumble_protocol::voice::{VoicePacket, VoicePacketPayload};
@@ -18,6 +18,8 @@ use tokio::net::UdpSocket;
 use tokio::sync::{mpsc, oneshot, watch};
 use tokio_util::udp::UdpFramed;
 
+pub type PingRequest = (u64, SocketAddr, Box<dyn FnOnce(PongPacket)>);
+
 type UdpSender = SplitSink<UdpFramed<ClientCryptState>, (VoicePacket<Serverbound>, SocketAddr)>;
 type UdpReceiver = SplitStream<UdpFramed<ClientCryptState>>;
 
@@ -90,17 +92,14 @@ async fn new_crypt_state(
     source: Arc<Mutex<UdpReceiver>>,
 ) {
     loop {
-        match crypt_state.recv().await {
-            Some(crypt_state) => {
-                info!("Received new crypt state");
-                let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16))
-                    .await
-                    .expect("Failed to bind UDP socket");
-                let (new_sink, new_source) = UdpFramed::new(udp_socket, crypt_state).split();
-                *sink.lock().unwrap() = new_sink;
-                *source.lock().unwrap() = new_source;
-            },
-            None => {},
+        if let Some(crypt_state) = crypt_state.recv().await {
+            info!("Received new crypt state");
+            let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16))
+                .await
+                .expect("Failed to bind UDP socket");
+            let (new_sink, new_source) = UdpFramed::new(udp_socket, crypt_state).split();
+            *sink.lock().unwrap() = new_sink;
+            *source.lock().unwrap() = new_source;
         }
     }
 }
@@ -248,11 +247,7 @@ async fn send_voice(
 }
 
 pub async fn handle_pings(
-    mut ping_request_receiver: mpsc::UnboundedReceiver<(
-        u64,
-        SocketAddr,
-        Box<dyn FnOnce(PongPacket)>,
-    )>,
+    mut ping_request_receiver: mpsc::UnboundedReceiver<PingRequest>,
 ) {
     let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16))
         .await
diff --git a/mumd/src/state.rs b/mumd/src/state.rs
index 85e5449..d1f64a9 100644
--- a/mumd/src/state.rs
+++ b/mumd/src/state.rs
@@ -57,17 +57,11 @@ pub struct State {
     server: Option<Server>,
     audio: Audio,
 
-    packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
-    connection_info_sender: watch::Sender<Option<ConnectionInfo>>,
-
     phase_watcher: (watch::Sender<StatePhase>, watch::Receiver<StatePhase>),
 }
 
 impl State {
-    pub fn new(
-        packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
-        connection_info_sender: watch::Sender<Option<ConnectionInfo>>,
-    ) -> Self {
+    pub fn new() -> Self {
         let config = mumlib::config::read_default_cfg();
         let audio = Audio::new(
             config.audio.input_volume.unwrap_or(1.0),
@@ -77,15 +71,18 @@ impl State {
             config,
             server: None,
             audio,
-            packet_sender,
-            connection_info_sender,
             phase_watcher: watch::channel(StatePhase::Disconnected),
         };
         state.reload_config();
         state
     }
 
-    pub fn handle_command(&mut self, command: Command) -> ExecutionContext {
+    pub fn handle_command(
+        &mut self,
+        command: Command,
+        packet_sender: &mut mpsc::UnboundedSender<ControlPacket<Serverbound>>,
+        connection_info_sender: &mut watch::Sender<Option<ConnectionInfo>>,
+    ) -> ExecutionContext {
         match command {
             Command::ChannelJoin { channel_identifier } => {
                 if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
@@ -134,7 +131,7 @@ impl State {
                 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();
+                packet_sender.send(msg.into()).unwrap();
                 now!(Ok(None))
             }
             Command::ChannelList => {
@@ -200,7 +197,7 @@ impl State {
                     let server = self.server_mut().unwrap();
                     server.set_muted(mute);
                     server.set_deafened(deafen);
-                    self.packet_sender.send(msg.into()).unwrap();
+                    packet_sender.send(msg.into()).unwrap();
                 }
 
                 now!(Ok(new_deaf.map(|b| CommandResponse::DeafenStatus { is_deafened: b })))
@@ -219,7 +216,7 @@ impl State {
                     .unwrap()
                     .users_mut()
                     .iter_mut()
-                    .find(|(_, user)| user.name() == &string);
+                    .find(|(_, user)| user.name() == string);
 
                 let (id, user) = match id {
                     Some(id) => (*id.0, id.1),
@@ -294,7 +291,7 @@ impl State {
                     let server = self.server_mut().unwrap();
                     server.set_muted(mute);
                     server.set_deafened(deafen);
-                    self.packet_sender.send(msg.into()).unwrap();
+                    packet_sender.send(msg.into()).unwrap();
                 }
 
                 now!(Ok(new_mute.map(|b| CommandResponse::MuteStatus { is_muted: b })))
@@ -334,7 +331,7 @@ impl State {
                         return now!(Err(Error::InvalidServerAddrError(host, port)));
                     }
                 };
-                self.connection_info_sender
+                connection_info_sender
                     .send(Some(ConnectionInfo::new(
                         socket_addr,
                         host,
@@ -408,7 +405,7 @@ impl State {
                     .unwrap()
                     .users()
                     .iter()
-                    .find(|e| e.1.name() == &string)
+                    .find(|e| e.1.name() == string)
                     .map(|e| *e.0)
                 {
                     None => return now!(Err(Error::InvalidUsernameError(string))),
@@ -592,9 +589,6 @@ impl State {
     pub fn audio_mut(&mut self) -> &mut Audio {
         &mut self.audio
     }
-    pub fn packet_sender(&self) -> mpsc::UnboundedSender<ControlPacket<Serverbound>> {
-        self.packet_sender.clone()
-    }
     pub fn phase_receiver(&self) -> watch::Receiver<StatePhase> {
         self.phase_watcher.1.clone()
     }
diff --git a/mumd/src/state/channel.rs b/mumd/src/state/channel.rs
index 8bbf919..5b6d669 100644
--- a/mumd/src/state/channel.rs
+++ b/mumd/src/state/channel.rs
@@ -88,7 +88,7 @@ impl<'a> ProtoTree<'a> {
                     users: Vec::new(),
                 });
                 pt.channel = Some(channel);
-                pt.users = users.get(&node).map(|e| e.clone()).unwrap_or(Vec::new());
+                pt.users = users.get(&node).cloned().unwrap_or_default();
             }
             longer => {
                 self.children
@@ -135,7 +135,7 @@ pub fn into_channel(
     for user in users.values() {
         channel_lookup
             .entry(user.channel())
-            .or_insert(Vec::new())
+            .or_insert_with(Vec::new)
             .push(user);
     }
 
@@ -148,7 +148,7 @@ pub fn into_channel(
         }
         walk.reverse();
 
-        if walk.len() > 0 {
+        if !walk.is_empty() {
             walks.push((walk, channel));
         }
     }
@@ -159,8 +159,8 @@ pub fn into_channel(
         children: HashMap::new(),
         users: channel_lookup
             .get(&0)
-            .map(|e| e.clone())
-            .unwrap_or(Vec::new()),
+            .cloned()
+            .unwrap_or_default(),
     };
 
     for (walk, channel) in walks {
diff --git a/mumd/src/state/server.rs b/mumd/src/state/server.rs
index a065df0..8a256b6 100644
--- a/mumd/src/state/server.rs
+++ b/mumd/src/state/server.rs
@@ -107,7 +107,7 @@ impl Server {
     }
 
     pub fn username(&self) -> Option<&str> {
-        self.username.as_ref().map(|e| e.as_str())
+        self.username.as_deref()
     }
 
     pub fn username_mut(&mut self) -> &mut Option<String> {