Handle muted & deafened state changes

src/protocol/parser.rs

@@ -155,11 +155,15 @@ pub struct UserRemove {
     pub session_id: SessionId,
 }
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct UserState {
     pub session_id: Option<SessionId>,
     pub name: Option<String>,
     pub channel_id: Option<u32>,
+    pub muted_by_admin: Option<bool>,
+    pub deafened_by_admin: Option<bool>,
+    pub self_mute: Option<bool>,
+    pub self_deaf: Option<bool>,
 }
 
 pub struct UdpTunnel {
@@ -183,6 +187,7 @@ impl ControlMessage {
             }
             CRYPT_SETUP => CryptSetup::from(mumble::CryptSetup::parse_from_bytes(payload)?).into(),
             PING => Ping::from(mumble::Ping::parse_from_bytes(payload)?).into(),
+            USER_STATE => UserState::from(mumble::UserState::parse_from_bytes(payload)?).into(),
             UDP_TUNNEL => ControlMessage::UdpTunnel(UdpTunnel {
                 audio_packet: AudioPacket::parse(payload.to_vec())?,
             }),
@@ -389,6 +394,10 @@ impl Message for UserState {
         let mut proto = mumble::UserState {
             name: SingularField::from(self.name),
             channel_id: self.channel_id,
+            mute: self.muted_by_admin,
+            deaf: self.deafened_by_admin,
+            self_mute: self.self_mute,
+            self_deaf: self.self_deaf,
             ..Default::default()
         };
         if let Some(session) = self.session_id {
@@ -520,14 +529,24 @@ impl From<mumble::Ping> for Ping {
     }
 }
 
+impl From<mumble::UserState> for UserState {
+    fn from(state: mumble::UserState) -> Self {
+        UserState {
+            session_id: state.session.map(SessionId::from),
+            name: state.name.into_option(),
+            channel_id: state.channel_id,
+            muted_by_admin: state.mute,
+            deafened_by_admin: state.deaf,
+            self_mute: state.self_mute,
+            self_deaf: state.self_deaf,
+        }
+    }
+}
+
 impl From<mumble::Version> for Version {
     fn from(version: mumble::Version) -> Self {
-        let mut protocol_version = None;
-        if let Some(version_number) = version.version {
-            protocol_version = Some(ProtocolVersion::from(version_number))
-        }
         Version {
-            version: protocol_version,
+            version: version.version.map(ProtocolVersion::from),
         }
     }
 }

src/server/client/client.rs

@@ -1,5 +1,5 @@
 use crate::protocol::connection::{AudioChannel, ControlChannel};
-use crate::protocol::parser::{AudioData, AudioPacket};
+use crate::protocol::parser::{AudioData, AudioPacket, UserState};
 use crate::server::client::handler::{Config, Error, Handler};
 use crate::storage::Storage;
 use std::marker::PhantomData;
@@ -18,12 +18,14 @@ pub struct Client<C: ControlChannel, A: AudioChannel> {
 
 pub enum ClientEvent {
     Talking(AudioData),
+    StateChanged(UserState),
     Disconnected,
 }
 
 pub enum ServerEvent {
     Connected(u32),
     Talking(AudioData),
+    StateChanged(UserState),
     Disconnected(u32),
 }
 

src/server/client/handler.rs

@@ -5,7 +5,7 @@ use crate::protocol::parser::{
     MUMBLE_PROTOCOL_VERSION,
 };
 use crate::server::client::client::{ClientEvent, ServerEvent};
-use crate::storage::{Guest, Storage};
+use crate::storage::{Guest, SessionData, Storage};
 use log::error;
 use ring::pbkdf2;
 use std::num::NonZeroU32;
@@ -154,8 +154,9 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
     pub async fn handle_server_event(&self, event: ServerEvent) -> Result<(), Error> {
         match event {
             ServerEvent::Connected(session_id) => self.new_user_connected(session_id).await?,
-            ServerEvent::Disconnected(session_id) => self.disconnected(session_id).await?,
-            ServerEvent::Talking(audio_data) => self.talking(audio_data).await?,
+            ServerEvent::StateChanged(state) => self.user_state_changed(state).await?,
+            ServerEvent::Talking(audio_data) => self.user_talking(audio_data).await?,
+            ServerEvent::Disconnected(session_id) => self.user_disconnected(session_id).await?,
         }
 
         Ok(())
@@ -164,6 +165,7 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
     pub async fn handle_message(&self, packet: ControlMessage) -> Result<(), Error> {
         match packet {
             ControlMessage::Ping(ping) => self.control_ping(ping).await?,
+            ControlMessage::UserState(state) => self.user_state(state).await?,
             ControlMessage::UdpTunnel(tunnel) => self.tunnel(tunnel).await?,
             _ => error!("unimplemented!"),
         }
@@ -212,6 +214,29 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
         Ok(())
     }
 
+    async fn user_state(&self, mut state: UserState) -> Result<(), Error> {
+        if state.session_id.is_none() {
+            state.session_id = Some(SessionId::from(self.session_id));
+        }
+
+        let session_data = SessionData {
+            muted_by_admin: state.muted_by_admin.unwrap_or_default(),
+            deafened_by_admin: state.deafened_by_admin.unwrap_or_default(),
+            suppressed: false,
+            self_mute: state.self_mute.unwrap_or_default(),
+            self_deaf: state.self_deaf.unwrap_or_default(),
+            priority_speaker: false,
+            recording: false,
+        };
+        self.storage
+            .update_session_data(self.session_id, session_data);
+        self.event_sender
+            .send(ClientEvent::StateChanged(state))
+            .await;
+
+        Ok(())
+    }
+
     async fn tunnel(&self, tunnel: UdpTunnel) -> Result<(), Error> {
         match tunnel.audio_packet {
             AudioPacket::Ping(_) => {
@@ -239,6 +264,7 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
                     session_id: id,
                     name: Some(user.username),
                     channel_id: Some(user.channel_id),
+                    ..Default::default()
                 })
                 .await?;
         } else if let Some(guest) = self.storage.get_guest(session_id) {
@@ -247,6 +273,7 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
                     session_id: id,
                     name: Some(guest.username),
                     channel_id: Some(guest.channel_id),
+                    ..Default::default()
                 })
                 .await?;
         }
@@ -254,7 +281,18 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
         Ok(())
     }
 
-    async fn talking(&self, audio_data: AudioData) -> Result<(), Error> {
+    async fn user_state_changed(&self, state: UserState) -> Result<(), Error> {
+        self.control_channel.send(state).await?;
+        Ok(())
+    }
+
+    async fn user_talking(&self, audio_data: AudioData) -> Result<(), Error> {
+        if let Some(data) = self.storage.get_session_data(self.session_id) {
+            if data.self_deaf || data.deafened_by_admin {
+                return Ok(());
+            }
+        }
+
         let audio_packet = AudioPacket::AudioData(audio_data);
         if let Some(channel) = self.audio_channel.as_ref() {
             channel.send(audio_packet).await?;
@@ -267,7 +305,7 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
         Ok(())
     }
 
-    async fn disconnected(&self, session_id: u32) -> Result<(), Error> {
+    async fn user_disconnected(&self, session_id: u32) -> Result<(), Error> {
         let user_remove = UserRemove {
             session_id: session_id.into(),
         };
@@ -330,6 +368,7 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
                 session_id: Some(SessionId::from(guest.session_id)),
                 name: Some(guest.username),
                 channel_id: Some(guest.channel_id),
+                ..Default::default()
             };
             states.push(state);
         }
@@ -339,6 +378,7 @@ impl<C: ControlChannel, A: AudioChannel> Handler<C, A> {
                 session_id: Some(SessionId::from(session_id)),
                 name: Some(user.username),
                 channel_id: Some(user.channel_id),
+                ..Default::default()
             };
             states.push(state);
         }

src/server/connection_worker.rs

@@ -1,5 +1,5 @@
 use crate::crypto::Ocb2Aes128Crypto;
-use crate::protocol::parser::AudioData;
+use crate::protocol::parser::{AudioData, UserState};
 use crate::server::client::{Client, ClientEvent, Config, Error, ServerEvent};
 use crate::server::session_pool::{SessionId, SessionPool};
 use crate::server::tcp_control_channel::TcpControlChannel;
@@ -91,28 +91,30 @@ impl ConnectionWorker {
             let message = event_receiver.recv().await.unwrap();
             match message {
                 ClientEvent::Disconnected => {
-                    self.client_disconnected().await;
+                    self.clients.remove(&self.session_id);
+                    self.broadcast_disconnect().await;
+                    self.session_pool.push(self.session_id);
                     return;
                 }
                 ClientEvent::Talking(audio_data) => {
-                    self.client_talking(audio_data).await;
+                    self.broadcast_audio(audio_data).await;
+                }
+                ClientEvent::StateChanged(state) => {
+                    self.broadcast_state_change(state).await;
                 }
             }
         }
     }
 
-    async fn client_disconnected(&self) {
-        self.clients.remove(&self.session_id);
+    async fn broadcast_disconnect(&self) {
         for client in self.clients.iter() {
             client
                 .send_event(ServerEvent::Disconnected(self.session_id))
                 .await;
         }
-
-        self.session_pool.push(self.session_id);
     }
 
-    async fn client_talking(&self, audio: AudioData) {
+    async fn broadcast_audio(&self, audio: AudioData) {
         for client in self
             .clients
             .iter()
@@ -121,4 +123,12 @@ impl ConnectionWorker {
             client.send_event(ServerEvent::Talking(audio.clone())).await;
         }
     }
+
+    async fn broadcast_state_change(&self, state: UserState) {
+        for client in self.clients.iter() {
+            client
+                .send_event(ServerEvent::StateChanged(state.clone()))
+                .await;
+        }
+    }
 }

src/storage.rs

@@ -59,13 +59,13 @@ pub struct Guest {
     pub texture_hash: Option<Sha1Hash>,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug, Default)]
 pub struct SessionData {
     pub muted_by_admin: bool,
     pub deafened_by_admin: bool,
     pub suppressed: bool,
-    pub muted: bool,
-    pub deafened: bool,
+    pub self_mute: bool,
+    pub self_deaf: bool,
     pub priority_speaker: bool,
     pub recording: bool,
 }
@@ -106,12 +106,15 @@ impl Storage {
     }
 
     pub fn add_guest(&self, guest: Guest) {
-        self.guests.insert(guest.session_id, guest);
+        let session_id = guest.session_id;
+        self.guests.insert(session_id, guest);
+        self.session_data.insert(session_id, SessionData::default());
     }
 
     pub fn add_connected_user(&self, user: User, session_id: SessionId) {
         self.connected_users
             .insert(session_id, (user.id, user.username));
+        self.session_data.insert(session_id, SessionData::default());
     }
 
     pub fn get_channels(&self) -> Vec<Channel> {
@@ -181,6 +184,10 @@ impl Storage {
             })
     }
 
+    pub fn update_session_data(&self, id: SessionId, data: SessionData) {
+        self.session_data.insert(id, data);
+    }
+
     pub fn username_in_connected(&self, username: &str) -> bool {
         if self
             .guests
@@ -228,8 +235,8 @@ impl SessionData {
             muted_by_admin: false,
             deafened_by_admin: false,
             suppressed: false,
-            muted: false,
-            deafened: false,
+            self_mute: false,
+            self_deaf: false,
             priority_speaker: false,
             recording: false,
         }