From 96ac028918baa1094374e823a2464016f7f20479 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Sat, 26 Dec 2020 22:33:10 +0100 Subject: add todos --- mumd/src/audio.rs | 13 ++++++------- mumd/src/network/tcp.rs | 8 ++++---- mumd/src/notify.rs | 2 +- 3 files changed, 11 insertions(+), 12 deletions(-) (limited to 'mumd') diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs index 6b46a7a..1e231e2 100644 --- a/mumd/src/audio.rs +++ b/mumd/src/audio.rs @@ -102,7 +102,7 @@ impl Audio { None } }) - .unwrap() + .unwrap() //TODO handle panic .with_sample_rate(sample_rate); let output_supported_sample_format = output_supported_config.sample_format(); let output_config: StreamConfig = output_supported_config.into(); @@ -120,7 +120,7 @@ impl Audio { None } }) - .unwrap() + .unwrap() //TODO handle panic .with_sample_rate(sample_rate); let input_supported_sample_format = input_supported_config.sample_format(); let input_config: StreamConfig = input_supported_config.into(); @@ -164,7 +164,7 @@ impl Audio { err_fn, ), } - .unwrap(); + .unwrap(); //TODO handle panic let (sample_sender, sample_receiver) = futures_channel::mpsc::channel(1_000_000); @@ -199,7 +199,7 @@ impl Audio { err_fn, ), } - .unwrap(); + .unwrap(); //TODO handle panic let opus_stream = OpusEncoder::new( 4, @@ -217,7 +217,7 @@ impl Audio { position_info: None, }); - output_stream.play().unwrap(); + output_stream.play().unwrap(); //TODO handle panic? let mut res = Self { output_config, @@ -268,7 +268,7 @@ impl Audio { let iter: Box> = match spec.channels { 1 => Box::new(samples.into_iter().flat_map(|e| vec![e, e])), 2 => Box::new(samples.into_iter()), - _ => unimplemented!() // TODO handle gracefully (this might not even happen) + _ => unimplemented!() // TODO handle panic (if speaker is surround speaker) }; let mut signal = signal::from_interleaved_samples_iter::<_, [f32; 2]>(iter); let interp = Linear::new(Signal::next(&mut signal), Signal::next(&mut signal)); @@ -401,4 +401,3 @@ fn get_sfx(file: &str) -> Cow<'static, [u8]> { fn get_default_sfx() -> Cow<'static, [u8]> { Cow::from(include_bytes!("fallback_sfx.wav").as_ref()) } - diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index 47b1c20..09cd844 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -177,7 +177,7 @@ async fn authenticate( msg.set_password(password); } msg.set_opus(true); - sink.send(msg.into()).await.unwrap(); + sink.send(msg.into()).await.unwrap(); //TODO handle panic } async fn send_pings( @@ -189,7 +189,7 @@ async fn send_pings( interval.tick().await; trace!("Sending TCP ping"); let msg = msgs::Ping::new(); - packet_sender.send(msg.into()).unwrap(); + packet_sender.send(msg.into()).unwrap(); //TODO handle panic } } @@ -198,8 +198,8 @@ async fn send_packets( packet_receiver: &mut mpsc::UnboundedReceiver>, ) { loop { - let packet = packet_receiver.recv().await.unwrap(); - sink.send(packet).await.unwrap(); + let packet = packet_receiver.recv().await.unwrap(); //TODO handle panic + sink.send(packet).await.unwrap(); //TODO handle panic } } diff --git a/mumd/src/notify.rs b/mumd/src/notify.rs index ee387cc..66a0faf 100644 --- a/mumd/src/notify.rs +++ b/mumd/src/notify.rs @@ -1,6 +1,6 @@ pub fn init() { #[cfg(feature = "notifications")] - libnotify::init("mumd").unwrap(); + libnotify::init("mumd").unwrap(); //TODO handle panic (don't send notifications) } #[cfg(feature = "notifications")] -- cgit v1.2.1 From e1907114374c842654f86b234b816f57dbbc79d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 11:21:38 +0200 Subject: add StateError and AudioError --- mumd/src/audio.rs | 23 ++++++++++++----------- mumd/src/client.rs | 3 ++- mumd/src/error.rs | 54 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ mumd/src/main.rs | 13 ++++++++++++- mumd/src/state.rs | 7 ++++--- 5 files changed, 84 insertions(+), 16 deletions(-) create mode 100644 mumd/src/error.rs (limited to 'mumd') diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs index 1e231e2..82acfee 100644 --- a/mumd/src/audio.rs +++ b/mumd/src/audio.rs @@ -4,6 +4,7 @@ mod noise_gate; use crate::audio::output::SaturatingAdd; use crate::audio::noise_gate::{from_interleaved_samples_stream, OpusEncoder, StreamingNoiseGate, StreamingSignalExt}; +use crate::error::{AudioError, AudioStream}; use crate::network::VoiceStreamType; use crate::state::StatePhase; @@ -85,16 +86,16 @@ pub struct Audio { } impl Audio { - pub fn new(input_volume: f32, output_volume: f32, phase_watcher: watch::Receiver) -> Self { + pub fn new(input_volume: f32, output_volume: f32, phase_watcher: watch::Receiver) -> Result { let sample_rate = SampleRate(SAMPLE_RATE); let host = cpal::default_host(); let output_device = host .default_output_device() - .expect("default output device not found"); + .ok_or(AudioError::NoDevice(AudioStream::Output))?; let output_supported_config = output_device .supported_output_configs() - .expect("error querying output configs") + .map_err(|e| AudioError::NoConfigs(AudioStream::Output, e))? .find_map(|c| { if c.min_sample_rate() <= sample_rate && c.max_sample_rate() >= sample_rate { Some(c) @@ -102,17 +103,17 @@ impl Audio { None } }) - .unwrap() //TODO handle panic + .ok_or(AudioError::NoSupportedConfig(AudioStream::Output))? .with_sample_rate(sample_rate); let output_supported_sample_format = output_supported_config.sample_format(); let output_config: StreamConfig = output_supported_config.into(); let input_device = host .default_input_device() - .expect("default input device not found"); + .ok_or(AudioError::NoDevice(AudioStream::Input))?; let input_supported_config = input_device .supported_input_configs() - .expect("error querying output configs") + .map_err(|e| AudioError::NoConfigs(AudioStream::Input, e))? .find_map(|c| { if c.min_sample_rate() <= sample_rate && c.max_sample_rate() >= sample_rate { Some(c) @@ -120,7 +121,7 @@ impl Audio { None } }) - .unwrap() //TODO handle panic + .ok_or(AudioError::NoSupportedConfig(AudioStream::Input))? .with_sample_rate(sample_rate); let input_supported_sample_format = input_supported_config.sample_format(); let input_config: StreamConfig = input_supported_config.into(); @@ -164,7 +165,7 @@ impl Audio { err_fn, ), } - .unwrap(); //TODO handle panic + .map_err(|e| AudioError::InvalidStream(AudioStream::Output, e))?; let (sample_sender, sample_receiver) = futures_channel::mpsc::channel(1_000_000); @@ -199,7 +200,7 @@ impl Audio { err_fn, ), } - .unwrap(); //TODO handle panic + .map_err(|e| AudioError::InvalidStream(AudioStream::Input, e))?; let opus_stream = OpusEncoder::new( 4, @@ -217,7 +218,7 @@ impl Audio { position_info: None, }); - output_stream.play().unwrap(); //TODO handle panic? + output_stream.play().map_err(|e| AudioError::OutputPlayError(e))?; let mut res = Self { output_config, @@ -232,7 +233,7 @@ impl Audio { play_sounds, }; res.load_sound_effects(&[]); - res + Ok(res) } pub fn load_sound_effects(&mut self, sound_effects: &[SoundEffect]) { diff --git a/mumd/src/client.rs b/mumd/src/client.rs index cdae7eb..6b66731 100644 --- a/mumd/src/client.rs +++ b/mumd/src/client.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use tokio::{join, sync::{Mutex, mpsc, oneshot, watch}}; pub async fn handle( + state: State, command_receiver: mpsc::UnboundedReceiver<( Command, oneshot::Sender>>, @@ -24,8 +25,8 @@ pub async fn handle( 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), diff --git a/mumd/src/error.rs b/mumd/src/error.rs new file mode 100644 index 0000000..a171f1f --- /dev/null +++ b/mumd/src/error.rs @@ -0,0 +1,54 @@ +use std::fmt; + +pub enum AudioStream { + Input, + Output, +} + +impl fmt::Display for AudioStream { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + AudioStream::Input => write!(f, "input"), + AudioStream::Output => write!(f, "output"), + } + } +} + +pub enum AudioError { + NoDevice(AudioStream), + NoConfigs(AudioStream, cpal::SupportedStreamConfigsError), + NoSupportedConfig(AudioStream), + InvalidStream(AudioStream, cpal::BuildStreamError), + OutputPlayError(cpal::PlayStreamError), +} + +impl fmt::Display for AudioError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + AudioError::NoDevice(s) => write!(f, "No {} device", s), + AudioError::NoConfigs(s, e) => write!(f, "No {} configs: {}", s, e), + AudioError::NoSupportedConfig(s) => write!(f, "No supported {} config found", s), + AudioError::InvalidStream(s, e) => write!(f, "Invalid {} stream: {}", s, e), + AudioError::OutputPlayError(e) => write!(f, "Playback error: {}", e), + } + } +} + +pub enum StateError { + AudioError(AudioError), +} + +impl From for StateError { + fn from(e: AudioError) -> Self { + StateError::AudioError(e) + } +} + +impl fmt::Display for StateError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + StateError::AudioError(e) => write!(f, "Audio error: {}", e), + } + } +} + diff --git a/mumd/src/main.rs b/mumd/src/main.rs index 276e2ce..42be3f8 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -1,10 +1,13 @@ mod audio; mod client; mod command; +mod error; mod network; mod notify; mod state; +use crate::state::State; + use futures_util::{SinkExt, StreamExt}; use log::*; use mumlib::command::{Command, CommandResponse}; @@ -53,8 +56,16 @@ async fn main() { let (command_sender, command_receiver) = mpsc::unbounded_channel(); + let state = match State::new() { + Ok(s) => s, + Err(e) => { + error!("Error instantiating mumd: {}", e); + return; + } + }; + join!( - client::handle(command_receiver), + client::handle(state, command_receiver), receive_commands(command_sender), ); } diff --git a/mumd/src/state.rs b/mumd/src/state.rs index 20fe660..858441a 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -3,6 +3,7 @@ pub mod server; pub mod user; use crate::audio::{Audio, NotificationEvents}; +use crate::error::StateError; use crate::network::{ConnectionInfo, VoiceStreamType}; use crate::network::tcp::{TcpEvent, TcpEventData}; use crate::notify; @@ -62,14 +63,14 @@ pub struct State { } impl State { - pub fn new() -> Self { + pub fn new() -> Result { let config = mumlib::config::read_default_cfg(); let phase_watcher = watch::channel(StatePhase::Disconnected); let audio = Audio::new( config.audio.input_volume.unwrap_or(1.0), config.audio.output_volume.unwrap_or(1.0), phase_watcher.1.clone(), - ); + ).map_err(|e| StateError::AudioError(e))?; let mut state = Self { config, server: None, @@ -77,7 +78,7 @@ impl State { phase_watcher, }; state.reload_config(); - state + Ok(state) } pub fn handle_command( -- cgit v1.2.1 From 0362df3a4a5d91cb2b8e530d44de327da7a18765 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 11:25:53 +0200 Subject: warn if notification init fails --- mumd/src/notify.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'mumd') diff --git a/mumd/src/notify.rs b/mumd/src/notify.rs index 66a0faf..e52b909 100644 --- a/mumd/src/notify.rs +++ b/mumd/src/notify.rs @@ -1,6 +1,10 @@ +use log::*; + pub fn init() { #[cfg(feature = "notifications")] - libnotify::init("mumd").unwrap(); //TODO handle panic (don't send notifications) + if libnotify::init("mumd").is_err() { + warn!("Unable to initialize notifications"); + } } #[cfg(feature = "notifications")] @@ -8,7 +12,7 @@ pub fn send(msg: String) -> Option { match libnotify::Notification::new("mumd", Some(msg.as_str()), None).show() { Ok(_) => Some(true), Err(_) => { - log::debug!("Unable to send notification"); + warn!("Unable to send notification"); Some(false) } } -- cgit v1.2.1 From d8c269f66a4a5a17155bec5e2cb5ad97f3ba00fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 11:27:10 +0200 Subject: rename notify -> notifications Notify is already taken, see https://lib.rs/crates/notify. --- mumd/src/main.rs | 4 ++-- mumd/src/notifications.rs | 24 ++++++++++++++++++++++++ mumd/src/notify.rs | 24 ------------------------ mumd/src/state.rs | 10 +++++----- 4 files changed, 31 insertions(+), 31 deletions(-) create mode 100644 mumd/src/notifications.rs delete mode 100644 mumd/src/notify.rs (limited to 'mumd') diff --git a/mumd/src/main.rs b/mumd/src/main.rs index 42be3f8..cd53d4a 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -3,7 +3,7 @@ mod client; mod command; mod error; mod network; -mod notify; +mod notifications; mod state; use crate::state::State; @@ -24,7 +24,7 @@ async fn main() { } setup_logger(std::io::stderr(), true); - notify::init(); + notifications::init(); // check if another instance is live let connection = UnixStream::connect(mumlib::SOCKET_PATH).await; diff --git a/mumd/src/notifications.rs b/mumd/src/notifications.rs new file mode 100644 index 0000000..e52b909 --- /dev/null +++ b/mumd/src/notifications.rs @@ -0,0 +1,24 @@ +use log::*; + +pub fn init() { + #[cfg(feature = "notifications")] + if libnotify::init("mumd").is_err() { + warn!("Unable to initialize notifications"); + } +} + +#[cfg(feature = "notifications")] +pub fn send(msg: String) -> Option { + match libnotify::Notification::new("mumd", Some(msg.as_str()), None).show() { + Ok(_) => Some(true), + Err(_) => { + warn!("Unable to send notification"); + Some(false) + } + } +} + +#[cfg(not(feature = "notifications"))] +pub fn send(_: String) -> Option { + None +} diff --git a/mumd/src/notify.rs b/mumd/src/notify.rs deleted file mode 100644 index e52b909..0000000 --- a/mumd/src/notify.rs +++ /dev/null @@ -1,24 +0,0 @@ -use log::*; - -pub fn init() { - #[cfg(feature = "notifications")] - if libnotify::init("mumd").is_err() { - warn!("Unable to initialize notifications"); - } -} - -#[cfg(feature = "notifications")] -pub fn send(msg: String) -> Option { - match libnotify::Notification::new("mumd", Some(msg.as_str()), None).show() { - Ok(_) => Some(true), - Err(_) => { - warn!("Unable to send notification"); - Some(false) - } - } -} - -#[cfg(not(feature = "notifications"))] -pub fn send(_: String) -> Option { - None -} diff --git a/mumd/src/state.rs b/mumd/src/state.rs index 858441a..9202e9f 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -6,7 +6,7 @@ use crate::audio::{Audio, NotificationEvents}; use crate::error::StateError; use crate::network::{ConnectionInfo, VoiceStreamType}; use crate::network::tcp::{TcpEvent, TcpEventData}; -use crate::notify; +use crate::notifications; use crate::state::server::Server; use log::*; @@ -463,7 +463,7 @@ impl State { == self.get_users_channel(self.server().unwrap().session_id().unwrap()) { if let Some(channel) = self.server().unwrap().channels().get(&channel_id) { - notify::send(format!( + notifications::send(format!( "{} connected and joined {}", &msg.get_name(), channel.name() @@ -516,7 +516,7 @@ impl State { self.get_users_channel(self.server().unwrap().session_id().unwrap()); if from_channel == this_channel || to_channel == this_channel { if let Some(channel) = self.server().unwrap().channels().get(&to_channel) { - notify::send(format!( + notifications::send(format!( "{} moved to channel {}", user.name(), channel.name() @@ -545,7 +545,7 @@ impl State { s += if deaf { " deafened" } else { " undeafened" }; } s += " themselves"; - notify::send(s); + notifications::send(s); } } } @@ -561,7 +561,7 @@ impl State { if this_channel == other_channel { self.audio.play_effect(NotificationEvents::UserDisconnected); if let Some(user) = self.server().unwrap().users().get(&msg.get_session()) { - notify::send(format!("{} disconnected", &user.name())); + notifications::send(format!("{} disconnected", &user.name())); } } -- cgit v1.2.1 From 950158eaadd8db9ef0eb48187e825524499422d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 12:36:53 +0200 Subject: config error --- mumd/src/error.rs | 9 +++++++++ mumd/src/state.rs | 9 +++++++-- 2 files changed, 16 insertions(+), 2 deletions(-) (limited to 'mumd') diff --git a/mumd/src/error.rs b/mumd/src/error.rs index a171f1f..84c1958 100644 --- a/mumd/src/error.rs +++ b/mumd/src/error.rs @@ -1,3 +1,4 @@ +use mumlib::error::ConfigError; use std::fmt; pub enum AudioStream { @@ -36,6 +37,7 @@ impl fmt::Display for AudioError { pub enum StateError { AudioError(AudioError), + ConfigError(ConfigError), } impl From for StateError { @@ -44,10 +46,17 @@ impl From for StateError { } } +impl From for StateError { + fn from(e: ConfigError) -> Self { + StateError::ConfigError(e) + } +} + impl fmt::Display for StateError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { StateError::AudioError(e) => write!(f, "Audio error: {}", e), + StateError::ConfigError(e) => write!(f, "Config error: {}", e), } } } diff --git a/mumd/src/state.rs b/mumd/src/state.rs index 9202e9f..b52b330 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -64,7 +64,7 @@ pub struct State { impl State { pub fn new() -> Result { - let config = mumlib::config::read_default_cfg(); + let config = mumlib::config::read_default_cfg()?; let phase_watcher = watch::channel(StatePhase::Disconnected); let audio = Audio::new( config.audio.input_volume.unwrap_or(1.0), @@ -574,7 +574,12 @@ impl State { } pub fn reload_config(&mut self) { - self.config = mumlib::config::read_default_cfg(); + match mumlib::config::read_default_cfg() { + Ok(config) => { + self.config = config; + } + Err(e) => error!("Couldn't read config: {}", e), + } if let Some(input_volume) = self.config.audio.input_volume { self.audio.set_input_volume(input_volume); } -- cgit v1.2.1 From 6bf6c40b8f6cf1fc08667a1049fc7f2ee4c0c140 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 12:48:55 +0200 Subject: skip handling >2 channels for now See #80. --- mumd/src/audio.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'mumd') diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs index 82acfee..fdbeaee 100644 --- a/mumd/src/audio.rs +++ b/mumd/src/audio.rs @@ -269,7 +269,7 @@ impl Audio { let iter: Box> = match spec.channels { 1 => Box::new(samples.into_iter().flat_map(|e| vec![e, e])), 2 => Box::new(samples.into_iter()), - _ => unimplemented!() // TODO handle panic (if speaker is surround speaker) + _ => unimplemented!("Only mono and stereo sound is supported. See #80.") }; let mut signal = signal::from_interleaved_samples_iter::<_, [f32; 2]>(iter); let interp = Linear::new(Signal::next(&mut signal), Signal::next(&mut signal)); -- cgit v1.2.1 From 80dba403ed968982ec23ab7416d48dc5b69329f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 14:18:18 +0200 Subject: return tcp errors from tcp internals --- mumd/src/error.rs | 23 +++++++++++++++++++++++ mumd/src/network/tcp.rs | 42 ++++++++++++++++++++++-------------------- 2 files changed, 45 insertions(+), 20 deletions(-) (limited to 'mumd') diff --git a/mumd/src/error.rs b/mumd/src/error.rs index 84c1958..6f77d52 100644 --- a/mumd/src/error.rs +++ b/mumd/src/error.rs @@ -1,5 +1,28 @@ +use mumble_protocol::{Serverbound, control::ControlPacket}; use mumlib::error::ConfigError; use std::fmt; +use tokio::sync::mpsc; + +pub enum TcpError { + NoConnectionInfoReceived, + TlsConnectorBuilderError(native_tls::Error), + TlsConnectError(native_tls::Error), + SendError(mpsc::error::SendError>), + + IOError(std::io::Error), +} + +impl From for TcpError { + fn from(e: std::io::Error) -> Self { + TcpError::IOError(e) + } +} + +impl From>> for TcpError { + fn from(e: mpsc::error::SendError>) -> Self { + TcpError::SendError(e) + } +} pub enum AudioStream { Input, diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index 09cd844..6460cba 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -1,3 +1,4 @@ +use crate::error::TcpError; use crate::network::ConnectionInfo; use crate::state::{State, StatePhase}; use log::*; @@ -84,7 +85,7 @@ pub async fn handle( packet_sender: mpsc::UnboundedSender>, mut packet_receiver: mpsc::UnboundedReceiver>, mut tcp_event_register_receiver: mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>, -) { +) -> Result<(), TcpError> { loop { let connection_info = 'data: loop { while connection_info_receiver.changed().await.is_ok() { @@ -92,20 +93,20 @@ pub async fn handle( break 'data data; } } - return; + return Err(TcpError::NoConnectionInfoReceived); }; let (mut sink, stream) = connect( connection_info.socket_addr, connection_info.hostname, connection_info.accept_invalid_cert, ) - .await; + .await?; // Handshake (omitting `Version` message for brevity) let state_lock = state.lock().await; let username = state_lock.username().unwrap().to_string(); let password = state_lock.password().map(|x| x.to_string()); - authenticate(&mut sink, username, password).await; + authenticate(&mut sink, username, password).await?; let phase_watcher = state_lock.phase_receiver(); let input_receiver = state_lock.audio().input_receiver(); drop(state_lock); @@ -115,6 +116,7 @@ pub async fn handle( run_until( |phase| matches!(phase, StatePhase::Disconnected), + //TODO take out the errors here and return them join5( send_pings(packet_sender.clone(), 10), listen( @@ -144,62 +146,62 @@ async fn connect( server_addr: SocketAddr, server_host: String, accept_invalid_cert: bool, -) -> (TcpSender, TcpReceiver) { - let stream = TcpStream::connect(&server_addr) - .await - .expect("failed to connect to server:"); +) -> Result<(TcpSender, TcpReceiver), TcpError> { + let stream = TcpStream::connect(&server_addr).await?; debug!("TCP connected"); let mut builder = native_tls::TlsConnector::builder(); builder.danger_accept_invalid_certs(accept_invalid_cert); let connector: TlsConnector = builder .build() - .expect("failed to create TLS connector") + .map_err(|e| TcpError::TlsConnectorBuilderError(e))? .into(); let tls_stream = connector .connect(&server_host, stream) .await - .expect("failed to connect TLS: {}"); + .map_err(|e| TcpError::TlsConnectError(e))?; debug!("TLS connected"); // Wrap the TLS stream with Mumble's client-side control-channel codec - ClientControlCodec::new().framed(tls_stream).split() + Ok(ClientControlCodec::new().framed(tls_stream).split()) } async fn authenticate( sink: &mut TcpSender, username: String, password: Option -) { +) -> Result<(), TcpError> { let mut msg = msgs::Authenticate::new(); msg.set_username(username); if let Some(password) = password { msg.set_password(password); } msg.set_opus(true); - sink.send(msg.into()).await.unwrap(); //TODO handle panic + sink.send(msg.into()).await?; + Ok(()) } async fn send_pings( packet_sender: mpsc::UnboundedSender>, delay_seconds: u64, -) { +) -> Result<(), TcpError> { let mut interval = time::interval(Duration::from_secs(delay_seconds)); loop { interval.tick().await; trace!("Sending TCP ping"); let msg = msgs::Ping::new(); - packet_sender.send(msg.into()).unwrap(); //TODO handle panic + packet_sender.send(msg.into())?; } } async fn send_packets( mut sink: TcpSender, packet_receiver: &mut mpsc::UnboundedReceiver>, -) { +) -> Result<(), TcpError> { loop { - let packet = packet_receiver.recv().await.unwrap(); //TODO handle panic - sink.send(packet).await.unwrap(); //TODO handle panic + // Safe since we always have at least one sender alive. + let packet = packet_receiver.recv().await.unwrap(); + sink.send(packet).await?; } } @@ -226,9 +228,9 @@ async fn send_voice( .await .next() .await - .unwrap() + .unwrap() //TODO handle panic .into()) - .unwrap(); + .unwrap(); //TODO handle panic } }, inner_phase_watcher.clone(), -- cgit v1.2.1 From 79702d18bbd23df2faf0c00b0d9537ce26508f6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 14:40:46 +0200 Subject: udp error --- mumd/src/client.rs | 3 ++- mumd/src/error.rs | 13 +++++++++++++ mumd/src/network/udp.rs | 16 ++++++++-------- 3 files changed, 23 insertions(+), 9 deletions(-) (limited to 'mumd') diff --git a/mumd/src/client.rs b/mumd/src/client.rs index 6b66731..7c1b0b7 100644 --- a/mumd/src/client.rs +++ b/mumd/src/client.rs @@ -27,7 +27,8 @@ pub async fn handle( let state = Arc::new(Mutex::new(state)); - join!( + //TODO report error here + let (_, _, _, _) = join!( tcp::handle( Arc::clone(&state), connection_info_receiver.clone(), diff --git a/mumd/src/error.rs b/mumd/src/error.rs index 6f77d52..e4a8fee 100644 --- a/mumd/src/error.rs +++ b/mumd/src/error.rs @@ -24,6 +24,19 @@ impl From>> for TcpError { } } +pub enum UdpError { + NoConnectionInfoReceived, + DisconnectBeforeCryptSetup, + + IOError(std::io::Error), +} + +impl From for UdpError { + fn from(e: std::io::Error) -> Self { + UdpError::IOError(e) + } +} + pub enum AudioStream { Input, Output, diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index da92dcb..5996e43 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -1,3 +1,4 @@ +use crate::error::UdpError; use crate::network::ConnectionInfo; use crate::state::{State, StatePhase}; @@ -31,7 +32,7 @@ pub async fn handle( state: Arc>, mut connection_info_receiver: watch::Receiver>, mut crypt_state_receiver: mpsc::Receiver, -) { +) -> Result<(), UdpError> { let receiver = state.lock().await.audio().input_receiver(); loop { @@ -41,9 +42,9 @@ pub async fn handle( break 'data data; } } - return; + return Err(UdpError::NoConnectionInfoReceived); }; - let (sink, source) = connect(&mut crypt_state_receiver).await; + let (sink, source) = connect(&mut crypt_state_receiver).await?; let sink = Arc::new(Mutex::new(sink)); let source = Arc::new(Mutex::new(source)); @@ -82,22 +83,21 @@ pub async fn handle( async fn connect( crypt_state: &mut mpsc::Receiver, -) -> (UdpSender, UdpReceiver) { +) -> Result<(UdpSender, UdpReceiver), UdpError> { // Bind UDP socket let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16)) - .await - .expect("Failed to bind UDP socket"); + .await?; // Wait for initial CryptState let crypt_state = match crypt_state.recv().await { Some(crypt_state) => crypt_state, // disconnected before we received the CryptSetup packet, oh well - None => panic!("Disconnect before crypt packet received"), //TODO exit gracefully + None => return Err(UdpError::DisconnectBeforeCryptSetup), }; debug!("UDP connected"); // Wrap the raw UDP packets in Mumble's crypto and voice codec (CryptState does both) - UdpFramed::new(udp_socket, crypt_state).split() + Ok(UdpFramed::new(udp_socket, crypt_state).split()) } async fn new_crypt_state( -- cgit v1.2.1 From 1d331f0707eaa4a056aa6261410fb1edb63097b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Tue, 30 Mar 2021 16:15:53 +0200 Subject: report tcp errors all the way --- mumd/src/client.rs | 25 +++++++++++++------------ mumd/src/error.rs | 28 +++++++++++++++++++++++++++- mumd/src/main.rs | 20 ++++++++++++++------ mumd/src/network.rs | 19 +++++++++++-------- mumd/src/network/tcp.rs | 46 +++++++++++++++++++++++++--------------------- 5 files changed, 90 insertions(+), 48 deletions(-) (limited to 'mumd') diff --git a/mumd/src/client.rs b/mumd/src/client.rs index 7c1b0b7..c1a0152 100644 --- a/mumd/src/client.rs +++ b/mumd/src/client.rs @@ -1,11 +1,13 @@ use crate::command; +use crate::error::ClientError; use crate::network::{tcp, udp, ConnectionInfo}; use crate::state::State; +use futures_util::{select, FutureExt}; use mumble_protocol::{Serverbound, control::ControlPacket, crypt::ClientCryptState}; use mumlib::command::{Command, CommandResponse}; use std::sync::Arc; -use tokio::{join, sync::{Mutex, mpsc, oneshot, watch}}; +use tokio::sync::{Mutex, mpsc, oneshot, watch}; pub async fn handle( state: State, @@ -13,7 +15,7 @@ pub async fn handle( Command, oneshot::Sender>>, )>, -) { +) -> Result<(), ClientError> { let (connection_info_sender, connection_info_receiver) = watch::channel::>(None); let (crypt_state_sender, crypt_state_receiver) = @@ -27,29 +29,28 @@ pub async fn handle( let state = Arc::new(Mutex::new(state)); - //TODO report error here - let (_, _, _, _) = join!( - tcp::handle( + select! { + r = tcp::handle( Arc::clone(&state), connection_info_receiver.clone(), crypt_state_sender, packet_sender.clone(), packet_receiver, response_receiver, - ), - udp::handle( + ).fuse() => r.map_err(|e| ClientError::TcpError(e)), + _ = udp::handle( Arc::clone(&state), connection_info_receiver.clone(), crypt_state_receiver, - ), - command::handle( + ).fuse() => Ok(()), + _ = command::handle( state, command_receiver, response_sender, ping_request_sender, packet_sender, connection_info_sender, - ), - udp::handle_pings(ping_request_receiver), - ); + ).fuse() => Ok(()), + _ = udp::handle_pings(ping_request_receiver).fuse() => Ok(()), + } } diff --git a/mumd/src/error.rs b/mumd/src/error.rs index e4a8fee..142e806 100644 --- a/mumd/src/error.rs +++ b/mumd/src/error.rs @@ -12,6 +12,21 @@ pub enum TcpError { IOError(std::io::Error), } +impl fmt::Display for TcpError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TcpError::NoConnectionInfoReceived + => write!(f, "No connection info received"), + TcpError::TlsConnectorBuilderError(e) + => write!(f, "Error building TLS connector: {}", e), + TcpError::TlsConnectError(e) + => write!(f, "TLS error when connecting: {}", e), + TcpError::SendError(e) => write!(f, "Couldn't send packet: {}", e), + TcpError::IOError(e) => write!(f, "IO error: {}", e), + } + } +} + impl From for TcpError { fn from(e: std::io::Error) -> Self { TcpError::IOError(e) @@ -37,6 +52,18 @@ impl From for UdpError { } } +pub enum ClientError { + TcpError(TcpError), +} + +impl fmt::Display for ClientError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ClientError::TcpError(e) => write!(f, "TCP error: {}", e), + } + } +} + pub enum AudioStream { Input, Output, @@ -96,4 +123,3 @@ impl fmt::Display for StateError { } } } - diff --git a/mumd/src/main.rs b/mumd/src/main.rs index cd53d4a..d7bc2c0 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -8,11 +8,11 @@ mod state; use crate::state::State; -use futures_util::{SinkExt, StreamExt}; +use futures_util::{select, FutureExt, SinkExt, StreamExt}; use log::*; use mumlib::command::{Command, CommandResponse}; use mumlib::setup_logger; -use tokio::{join, net::{UnixListener, UnixStream}, sync::{mpsc, oneshot}}; +use tokio::{net::{UnixListener, UnixStream}, sync::{mpsc, oneshot}}; use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec}; use bytes::{BufMut, BytesMut}; @@ -64,10 +64,18 @@ async fn main() { } }; - join!( - client::handle(state, command_receiver), - receive_commands(command_sender), - ); + let run = select! { + r = client::handle(state, command_receiver).fuse() => r, + _ = receive_commands(command_sender).fuse() => Ok(()), + }; + + match run { + Err(e) => { + error!("mumd: {}", e); + std::process::exit(1); + } + _ => {} + } } async fn receive_commands( diff --git a/mumd/src/network.rs b/mumd/src/network.rs index 38a97ce..4eca90d 100644 --- a/mumd/src/network.rs +++ b/mumd/src/network.rs @@ -4,7 +4,7 @@ pub mod udp; use futures_util::FutureExt; use log::*; use std::{future::Future, net::SocketAddr}; -use tokio::{join, select, sync::{oneshot, watch}}; +use tokio::{select, sync::{oneshot, watch}}; use crate::state::StatePhase; @@ -31,12 +31,12 @@ pub enum VoiceStreamType { UDP, } -async fn run_until( +async fn run_until( phase_checker: impl Fn(StatePhase) -> bool, fut: F, mut phase_watcher: watch::Receiver, -) where - F: Future, +) -> Option + where F: Future, { let (tx, rx) = oneshot::channel(); let phase_transition_block = async { @@ -55,10 +55,13 @@ async fn run_until( let rx = rx.fuse(); let fut = fut.fuse(); select! { - _ = fut => (), - _ = rx => (), - }; + r = fut => Some(r), + _ = rx => None, + } }; - join!(main_block, phase_transition_block); + select! { + m = main_block => m, + _ = phase_transition_block => None, + } } diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index 6460cba..9b0b68e 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -4,6 +4,7 @@ use crate::state::{State, StatePhase}; use log::*; use futures_util::{FutureExt, SinkExt, StreamExt}; +use futures_util::select; use futures_util::stream::{SplitSink, SplitStream, Stream}; use mumble_protocol::control::{msgs, ClientControlCodec, ControlCodec, ControlPacket}; use mumble_protocol::crypt::ClientCryptState; @@ -20,7 +21,6 @@ use tokio_native_tls::{TlsConnector, TlsStream}; use tokio_util::codec::{Decoder, Framed}; use super::{run_until, VoiceStreamType}; -use futures_util::future::join5; type TcpSender = SplitSink< Framed, ControlCodec>, @@ -114,27 +114,30 @@ pub async fn handle( info!("Logging in..."); + let phase_watcher_inner = phase_watcher.clone(); + run_until( |phase| matches!(phase, StatePhase::Disconnected), - //TODO take out the errors here and return them - join5( - send_pings(packet_sender.clone(), 10), - listen( - Arc::clone(&state), - stream, - crypt_state_sender.clone(), - event_queue.clone(), - ), - send_voice( - packet_sender.clone(), - Arc::clone(&input_receiver), - phase_watcher.clone(), - ), - send_packets(sink, &mut packet_receiver), - register_events(&mut tcp_event_register_receiver, event_queue.clone()), - ).map(|_| ()), + async { + select! { + r = send_pings(packet_sender.clone(), 10).fuse() => r, + r = listen( + Arc::clone(&state), + stream, + crypt_state_sender.clone(), + event_queue.clone(), + ).fuse() => r, + r = send_voice( + packet_sender.clone(), + Arc::clone(&input_receiver), + 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, - ).await; + ).await.unwrap_or(Ok(()))?; event_queue.resolve(TcpEventData::Disconnected).await; @@ -209,7 +212,7 @@ async fn send_voice( packet_sender: mpsc::UnboundedSender>, receiver: Arc> + Unpin)>>>, phase_watcher: watch::Receiver, -) { +) -> Result<(), TcpError> { loop { let mut inner_phase_watcher = phase_watcher.clone(); loop { @@ -243,7 +246,7 @@ async fn listen( mut stream: TcpReceiver, crypt_state_sender: mpsc::Sender, event_queue: TcpEventQueue, -) { +) -> Result<(), TcpError> { let mut crypt_state = None; let mut crypt_state_sender = Some(crypt_state_sender); @@ -369,6 +372,7 @@ async fn listen( } } } + Ok(()) } async fn register_events( -- cgit v1.2.1 From 69f189fd45b410be2db3c77e2a4bfa6d9ad8946d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= Date: Wed, 31 Mar 2021 09:05:12 +0200 Subject: review --- mumd/src/error.rs | 8 +++++--- mumd/src/network/tcp.rs | 14 ++++++++------ mumd/src/notifications.rs | 8 ++++---- 3 files changed, 17 insertions(+), 13 deletions(-) (limited to 'mumd') diff --git a/mumd/src/error.rs b/mumd/src/error.rs index 142e806..f7818a1 100644 --- a/mumd/src/error.rs +++ b/mumd/src/error.rs @@ -3,11 +3,13 @@ use mumlib::error::ConfigError; use std::fmt; use tokio::sync::mpsc; +pub type ServerSendError = mpsc::error::SendError>; + pub enum TcpError { NoConnectionInfoReceived, TlsConnectorBuilderError(native_tls::Error), TlsConnectError(native_tls::Error), - SendError(mpsc::error::SendError>), + SendError(ServerSendError), IOError(std::io::Error), } @@ -33,8 +35,8 @@ impl From for TcpError { } } -impl From>> for TcpError { - fn from(e: mpsc::error::SendError>) -> Self { +impl From for TcpError { + fn from(e: ServerSendError) -> Self { TcpError::SendError(e) } } diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index 9b0b68e..5783cc8 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -1,4 +1,4 @@ -use crate::error::TcpError; +use crate::error::{ServerSendError, TcpError}; use crate::network::ConnectionInfo; use crate::state::{State, StatePhase}; use log::*; @@ -225,19 +225,21 @@ async fn send_voice( |phase| !matches!(phase, StatePhase::Connected(VoiceStreamType::TCP)), async { loop { - packet_sender.send( + let res: Result<(), ServerSendError> = packet_sender.send( receiver .lock() .await .next() .await - .unwrap() //TODO handle panic - .into()) - .unwrap(); //TODO handle panic + .expect("No audio stream") + .into()); + if matches!(res, Err(_)) { + return res; + } } }, inner_phase_watcher.clone(), - ).await; + ).await.unwrap_or(Ok(()))?; } } diff --git a/mumd/src/notifications.rs b/mumd/src/notifications.rs index e52b909..bccf4dd 100644 --- a/mumd/src/notifications.rs +++ b/mumd/src/notifications.rs @@ -2,8 +2,8 @@ use log::*; pub fn init() { #[cfg(feature = "notifications")] - if libnotify::init("mumd").is_err() { - warn!("Unable to initialize notifications"); + if let Err(e) = libnotify::init("mumd") { + warn!("Unable to initialize notifications: {}", e); } } @@ -11,8 +11,8 @@ pub fn init() { pub fn send(msg: String) -> Option { match libnotify::Notification::new("mumd", Some(msg.as_str()), None).show() { Ok(_) => Some(true), - Err(_) => { - warn!("Unable to send notification"); + Err(e) => { + warn!("Unable to send notification: {}", e); Some(false) } } -- cgit v1.2.1 From 46a3938b6d9d81649e38e6e793599a52991d803d Mon Sep 17 00:00:00 2001 From: Eskil Queseth Date: Wed, 31 Mar 2021 21:50:50 +0200 Subject: tyrbofish ? --- mumd/src/network/tcp.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) (limited to 'mumd') diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index 5783cc8..6402a89 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -225,21 +225,18 @@ async fn send_voice( |phase| !matches!(phase, StatePhase::Connected(VoiceStreamType::TCP)), async { loop { - let res: Result<(), ServerSendError> = packet_sender.send( + packet_sender.send( receiver .lock() .await .next() .await .expect("No audio stream") - .into()); - if matches!(res, Err(_)) { - return res; - } + .into())?; } }, inner_phase_watcher.clone(), - ).await.unwrap_or(Ok(()))?; + ).await.unwrap_or(Ok::<(), ServerSendError>(()))?; } } -- cgit v1.2.1