add voice stream type

5 files changed, 53 insertions, 38 deletions

diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs
index 680433c..40cdcb2 100644
--- a/mumd/src/audio.rs
+++ b/mumd/src/audio.rs
@@ -2,6 +2,7 @@ pub mod input;
 pub mod output;
 
 use crate::audio::output::SaturatingAdd;
+use crate::network::VoiceStreamType;
 
 use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 use cpal::{SampleFormat, SampleRate, StreamConfig};
@@ -82,7 +83,7 @@ pub struct Audio {
 
     user_volumes: Arc<Mutex<HashMap<u32, (f32, bool)>>>,
 
-    client_streams: Arc<Mutex<HashMap<u32, output::ClientStream>>>,
+    client_streams: Arc<Mutex<HashMap<(VoiceStreamType, u32), output::ClientStream>>>,
 
     sounds: HashMap<NotificationEvents, Vec<f32>>,
     play_sounds: Arc<Mutex<VecDeque<f32>>>,
@@ -291,8 +292,8 @@ impl Audio {
             .collect();
     }
 
-    pub fn decode_packet(&self, session_id: u32, payload: VoicePacketPayload) {
-        match self.client_streams.lock().unwrap().entry(session_id) {
+    pub fn decode_packet(&self, stream_type: VoiceStreamType, session_id: u32, payload: VoicePacketPayload) {
+        match self.client_streams.lock().unwrap().entry((stream_type, session_id)) {
             Entry::Occupied(mut entry) => {
                 entry
                     .get_mut()
@@ -305,29 +306,33 @@ impl Audio {
     }
 
     pub fn add_client(&self, session_id: u32) {
-        match self.client_streams.lock().unwrap().entry(session_id) {
-            Entry::Occupied(_) => {
-                warn!("Session id {} already exists", session_id);
-            }
-            Entry::Vacant(entry) => {
-                entry.insert(output::ClientStream::new(
-                    self.output_config.sample_rate.0,
-                    self.output_config.channels,
-                ));
+        for stream_type in [VoiceStreamType::TCP, VoiceStreamType::UDP].iter() {
+            match self.client_streams.lock().unwrap().entry((*stream_type, session_id)) {
+                Entry::Occupied(_) => {
+                    warn!("Session id {} already exists", session_id);
+                }
+                Entry::Vacant(entry) => {
+                    entry.insert(output::ClientStream::new(
+                        self.output_config.sample_rate.0,
+                        self.output_config.channels,
+                    ));
+                }
             }
         }
     }
 
     pub fn remove_client(&self, session_id: u32) {
-        match self.client_streams.lock().unwrap().entry(session_id) {
-            Entry::Occupied(entry) => {
-                entry.remove();
-            }
-            Entry::Vacant(_) => {
-                warn!(
-                    "Tried to remove session id {} that doesn't exist",
-                    session_id
-                );
+        for stream_type in [VoiceStreamType::TCP, VoiceStreamType::UDP].iter() {
+            match self.client_streams.lock().unwrap().entry((*stream_type, session_id)) {
+                Entry::Occupied(entry) => {
+                    entry.remove();
+                }
+                Entry::Vacant(_) => {
+                    warn!(
+                        "Tried to remove session id {} that doesn't exist",
+                        session_id
+                    );
+                }
             }
         }
     }
diff --git a/mumd/src/audio/output.rs b/mumd/src/audio/output.rs
index 5e0cb8d..421d395 100644
--- a/mumd/src/audio/output.rs
+++ b/mumd/src/audio/output.rs
@@ -1,3 +1,5 @@
+use crate::network::VoiceStreamType;
+
 use cpal::{OutputCallbackInfo, Sample};
 use mumble_protocol::voice::VoicePacketPayload;
 use opus::Channels;
@@ -73,7 +75,7 @@ impl SaturatingAdd for u16 {
 
 pub fn curry_callback<T: Sample + AddAssign + SaturatingAdd + std::fmt::Display>(
     effect_sound: Arc<Mutex<VecDeque<f32>>>,
-    user_bufs: Arc<Mutex<HashMap<u32, ClientStream>>>,
+    user_bufs: Arc<Mutex<HashMap<(VoiceStreamType, u32), ClientStream>>>,
     output_volume_receiver: watch::Receiver<f32>,
     user_volumes: Arc<Mutex<HashMap<u32, (f32, bool)>>>,
 ) -> impl FnMut(&mut [T], &OutputCallbackInfo) + Send + 'static {
@@ -86,7 +88,7 @@ pub fn curry_callback<T: Sample + AddAssign + SaturatingAdd + std::fmt::Display>
 
         let mut effects_sound = effect_sound.lock().unwrap();
         let mut user_bufs = user_bufs.lock().unwrap();
-        for (id, client_stream) in &mut *user_bufs {
+        for ((_, id), client_stream) in &mut *user_bufs {
             let (user_volume, muted) = user_volumes
                 .lock()
                 .unwrap()
diff --git a/mumd/src/network.rs b/mumd/src/network.rs
index 1a31ee2..4fb2e77 100644
--- a/mumd/src/network.rs
+++ b/mumd/src/network.rs
@@ -19,3 +19,9 @@ impl ConnectionInfo {
         }
     }
 }
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
+pub enum VoiceStreamType {
+    TCP,
+    UDP,
+}
diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs
index 0c00029..1465e8c 100644
--- a/mumd/src/network/udp.rs
+++ b/mumd/src/network/udp.rs
@@ -18,6 +18,8 @@ use tokio::net::UdpSocket;
 use tokio::sync::{mpsc, oneshot, watch};
 use tokio_util::udp::UdpFramed;
 
+use super::VoiceStreamType;
+
 pub type PingRequest = (u64, SocketAddr, Box<dyn FnOnce(PongPacket)>);
 
 type UdpSender = SplitSink<UdpFramed<ClientCryptState>, (VoicePacket<Serverbound>, SocketAddr)>;
@@ -165,7 +167,7 @@ async fn listen(
                                 .lock()
                                 .unwrap()
                                 .audio()
-                                .decode_packet(session_id, payload);
+                                .decode_packet(VoiceStreamType::UDP, session_id, payload);
                         }
                     }
                 }
diff --git a/mumd/src/state.rs b/mumd/src/state.rs
index 84247bc..4e8a886 100644
--- a/mumd/src/state.rs
+++ b/mumd/src/state.rs
@@ -3,11 +3,11 @@ pub mod server;
 pub mod user;
 
 use crate::audio::{Audio, NotificationEvents};
-use crate::network::ConnectionInfo;
+use crate::network::{ConnectionInfo, VoiceStreamType};
+use crate::network::tcp::{TcpEvent, TcpEventData};
 use crate::notify;
 use crate::state::server::Server;
 
-use crate::network::tcp::{TcpEvent, TcpEventData};
 use log::*;
 use mumble_protocol::control::msgs;
 use mumble_protocol::control::ControlPacket;
@@ -45,11 +45,11 @@ pub enum ExecutionContext {
     ),
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum StatePhase {
     Disconnected,
     Connecting,
-    Connected,
+    Connected(VoiceStreamType),
 }
 
 pub struct State {
@@ -85,7 +85,7 @@ impl State {
     ) -> ExecutionContext {
         match command {
             Command::ChannelJoin { channel_identifier } => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
 
@@ -135,7 +135,7 @@ impl State {
                 now!(Ok(None))
             }
             Command::ChannelList => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
                 let list = channel::into_channel(
@@ -149,7 +149,7 @@ impl State {
                 now!(Ok(None))
             }
             Command::DeafenSelf(toggle) => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
 
@@ -207,7 +207,7 @@ impl State {
                 now!(Ok(None))
             }
             Command::MuteOther(string, toggle) => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
 
@@ -242,7 +242,7 @@ impl State {
                 return now!(Ok(None));
             }
             Command::MuteSelf(toggle) => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
 
@@ -354,7 +354,7 @@ impl State {
                 })
             }
             Command::ServerDisconnect => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
 
@@ -388,7 +388,7 @@ impl State {
                 }),
             ),
             Command::Status => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
                 let state = self.server.as_ref().unwrap().into();
@@ -397,7 +397,7 @@ impl State {
                 })))
             }
             Command::UserVolumeSet(string, volume) => {
-                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) {
+                if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                     return now!(Err(Error::DisconnectedError));
                 }
                 let user_id = match self
@@ -448,7 +448,7 @@ impl State {
             self.audio_mut().add_client(session);
 
             // send notification only if we've passed the connecting phase
-            if *self.phase_receiver().borrow() == StatePhase::Connected {
+            if matches!(*self.phase_receiver().borrow(), StatePhase::Connected(_)) {
                 let channel_id = msg.get_channel_id();
 
                 if channel_id
@@ -581,7 +581,7 @@ impl State {
     pub fn initialized(&self) {
         self.phase_watcher
             .0
-            .send(StatePhase::Connected)
+            .send(StatePhase::Connected(VoiceStreamType::UDP))
             .unwrap();
         self.audio.play_effect(NotificationEvents::ServerConnect);
     }