diff --git a/readme.md b/readme.md index 8803f60..2c41f5d 100644 --- a/readme.md +++ b/readme.md @@ -13,6 +13,7 @@ - Figure out Meter level encoding - Full test of client (Improve existing one) - Write correct readme.md +- Make client dsp408 thread safe Rust library for remote LAN control of Thomann DSP408 digital signal processors. diff --git a/src/common/errors.rs b/src/common/errors.rs index 8587e8e..0c92a2d 100644 --- a/src/common/errors.rs +++ b/src/common/errors.rs @@ -122,6 +122,12 @@ pub enum DSPError { #[error("operation failed: {0}")] OperationFailed(String), + #[error("internal error: {0}")] + InternalError(String), + + #[error("mutex poisoned: {0}")] + MutexPoisoned(String), + #[error("I/O error: {0}")] Io(#[from] std::io::Error), @@ -134,3 +140,9 @@ pub enum DSPError { #[error("frame error: {0}")] Frame(#[from] FrameError), } + +impl From> for DSPError { + fn from(err: std::sync::PoisonError) -> Self { + DSPError::MutexPoisoned(err.to_string()) + } +} diff --git a/src/dsp408/client/dsp408.rs b/src/dsp408/client/dsp408.rs index 411e1c4..66841f7 100644 --- a/src/dsp408/client/dsp408.rs +++ b/src/dsp408/client/dsp408.rs @@ -32,8 +32,8 @@ fn instances() -> &'static Mutex> { } pub struct DSP408 { - connection: types::DeviceConnection, - socket: Arc>>, + connection: types::DeviceConnection, // read-only so thread protected + socket: Arc>>, // thread protected connected: bool, state: Option, } @@ -93,7 +93,10 @@ impl DSP408 { } pub fn send_recv(&self, command: &[u8]) -> Result { - let mut socket_guard = self.socket.lock().unwrap(); + let mut socket_guard = self + .socket + .lock() + .map_err(|e| DSPError::MutexPoisoned(e.to_string()))?; let socket = socket_guard.as_mut().ok_or(DSPError::NotConnected)?; @@ -124,13 +127,42 @@ impl DSP408 { self.state.as_mut().ok_or(DSPError::NotConnected) } + fn initialize_state(&mut self) -> Result<(), DSPError> { + let name = self.get_device_name()?; + let flags = self.get_device_flags()?; + let current_preset = self.get_current_preset()?; + let modified = self.get_preset_modification_status()?; + let mut preset_names = Vec::with_capacity(PRESET_INDEX_MAX); + + for i in 0..PRESET_INDEX_MAX { + preset_names.push(self.get_preset_name(i + 1)?); + } + + let state = self.read_config_state()?; + + self.state = Some(types::DSPState { + name, + flags, + presets: types::PresetBank { + current_index: current_preset as usize, + names: preset_names, + modified: modified.modified, + }, + current_config: state, + }); + + Ok(()) + } + // ---------- Commands ---------- // Command `0x10` pub fn connect(&mut self) -> Result<(), DSPError> { let key = (self.connection.host, self.connection.device_id); { - let map = instances().lock().unwrap(); + let map = instances() + .lock() + .map_err(|e| DSPError::InternalError(e.to_string()))?; if map.contains_key(&key) { return Err(DSPError::AlreadyConnected); @@ -144,14 +176,21 @@ impl DSP408 { stream.set_read_timeout(Some(SOCKET_TIMEOUT))?; { - let mut socket = self.socket.lock().unwrap(); + let mut socket = self + .socket + .lock() + .map_err(|e| DSPError::InternalError(e.to_string()))?; + *socket = Some(stream); } let handshake = commands::build_handshake(); { - let mut socket = self.socket.lock().unwrap(); + let mut socket = self + .socket + .lock() + .map_err(|e| DSPError::InternalError(e.to_string()))?; let socket = socket.as_mut().ok_or(DSPError::NotConnected)?; @@ -162,66 +201,64 @@ impl DSP408 { while start.elapsed() < SOCKET_TIMEOUT { let raw = { - let mut socket = self.socket.lock().unwrap(); + let mut socket = self + .socket + .lock() + .map_err(|e| DSPError::InternalError(e.to_string()))?; let socket = socket.as_mut().ok_or(DSPError::NotConnected)?; Self::recv_raw(socket)? }; + let frame = extract_frame(&raw)?; let response = map_response(&frame)?; if matches!(response, types::DeviceResponseType::HandshakeAck(_)) { - instances().lock().unwrap().insert(key, ()); + let result = self.initialize_state(); - self.connected = true; - - let name = self.get_device_name()?; - let flags = self.get_device_flags()?; - let current_preset = self.get_current_preset()?; - let modified = self.get_preset_modification_status()?; - let mut preset_names = Vec::with_capacity(PRESET_INDEX_MAX); - - for i in 0..PRESET_INDEX_MAX { - preset_names.push(self.get_preset_name(i + 1)?); + if result.is_err() { + self.disconnect()?; + return result; } - let state = self.read_config_state()?; + instances() + .lock() + .map_err(|e| DSPError::InternalError(e.to_string()))? + .insert(key, ()); - self.state = Some(types::DSPState { - name, - flags, - presets: types::PresetBank { - current_index: current_preset as usize, - names: preset_names, - modified: modified.modified, - }, - current_config: state, - }); + self.connected = true; return Ok(()); } } - self.disconnect(); + self.disconnect()?; Err(DSPError::HandshakeFailed) } // Command `0x11` - pub fn disconnect(&mut self) { + pub fn disconnect(&mut self) -> Result<(), DSPError> { let key = (self.connection.host, self.connection.device_id); - let mut socket_guard = self.socket.lock().unwrap(); + let mut socket_guard = self.socket.lock()?; if let Some(mut socket) = socket_guard.take() { - let _ = socket.write_all(&commands::build_disconnect()); - let _ = socket.shutdown(std::net::Shutdown::Both); + socket + .write_all(&commands::build_disconnect()) + .map_err(DSPError::Io)?; + + socket + .shutdown(std::net::Shutdown::Both) + .map_err(DSPError::Io)?; } - instances().lock().unwrap().remove(&key); + instances().lock()?.remove(&key); self.connected = false; + + Ok(()) } // Command `0x12` @@ -1428,7 +1465,9 @@ impl DSP408 { impl Drop for DSP408 { fn drop(&mut self) { - self.disconnect(); + if self.connected { + let _ = self.disconnect(); + } } } diff --git a/src/main.rs b/src/main.rs index c3459ca..7666a42 100644 --- a/src/main.rs +++ b/src/main.rs @@ -27,7 +27,7 @@ fn main() -> Result<(), Box> { dbg!(dsp.state()?); println!("Disconnecting..."); - dsp.disconnect(); + dsp.disconnect()?; println!("Disconnected"); Ok(())