//! Native Vivox gateway using the bounded TCP control protocol. use crate::*; use base64::Engine as _; use libremetaverse_types::compat::{EventHandler, Subscription}; use libremetaverse_types::{UUID, Vector3d}; use roxmltree::{Document, Node}; use std::any::Any; use std::collections::HashMap; use std::sync::{ Arc, Mutex, Weak, atomic::{AtomicBool, AtomicU64, Ordering}, }; const MAX_XML_BYTES: usize = 1024 * 1024; fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { mutex .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) } #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] enum EventKey { AccountLoginResponse, AccountLoginState, AuxAudioProperties, CaptureDevices, RenderDevices, ConnectorCreate, DaemonConnected, DaemonCouldntConnect, DaemonCouldntRun, DaemonDisconnected, DaemonExited, DaemonRunning, SessionAdded, SessionCreate, SessionCreateResponse, SessionGroupAdded, SessionMedia, SessionNew, ParticipantAdded, ParticipantProperties, ParticipantRemoved, ParticipantState, ParticipantUpdated, SessionRemove, SessionRemoved, SessionState, SessionUpdated, VoiceConnection, VoiceMicTest, VoiceResponse, RawControlLine, } type ErasedHandler = Arc; #[derive(Default)] struct EventBus { next_id: AtomicU64, handlers: Arc>>>, } impl EventBus { fn subscribe(&self, key: EventKey, handler: Option>) -> Subscription where T: Any + Clone + Send + Sync + 'static, { let Some(handler) = handler else { return Subscription::detached(); }; let id = self.next_id.fetch_add(1, Ordering::Relaxed); let erased: ErasedHandler = Arc::new(move |value| { if let Some(value) = value.downcast_ref::() { handler(value.clone()); } }); lock(&self.handlers) .entry(key) .or_default() .insert(id, erased); let handlers = Arc::downgrade(&self.handlers); Subscription::new(move || remove_handler(&handlers, key, id)) } fn emit(&self, key: EventKey, value: T) where T: Any + Send + Sync + 'static, { let handlers: Vec<_> = lock(&self.handlers) .get(&key) .into_iter() .flat_map(|handlers| handlers.values().cloned()) .collect(); for handler in handlers { handler(&value); } } } fn remove_handler( handlers: &Weak>>>, key: EventKey, id: u64, ) { let Some(handlers) = handlers.upgrade() else { return; }; let mut handlers = lock(&handlers); if let Some(event) = handlers.get_mut(&key) { event.remove(&id); if event.is_empty() { handlers.remove(&key); } } } #[derive(Default)] struct GatewayState { active: bool, current_capture_device: String, current_playback_device: String, daemon_connected: bool, daemon_running: bool, request_id: i32, testing: bool, connector_handle: String, account_handle: String, session_handle: String, sip_server: String, sessions: HashMap, } struct GatewayInner { _client: libremetaverse::GridClient, pipe: TCPPipe, state: Mutex, events: EventBus, pipe_subscriptions: Mutex>, disposed: AtomicBool, } #[derive(Clone)] pub struct VoiceGateway { inner: Arc, } macro_rules! event_subscription { ($name:ident, $key:ident, $type:ty) => { #[must_use] pub fn $name(&self, handler: Option>) -> Subscription { self.inner.events.subscribe(EventKey::$key, handler) } }; } macro_rules! zero_callback_subscription { ($name:ident, $key:ident, $type:ty) => { #[must_use] pub fn $name(&self, handler: Option<$type>) -> Subscription { let handler = handler.map(|handler| { Arc::new(move |()| { let _ = handler.invoke(); }) as EventHandler<()> }); self.inner.events.subscribe(EventKey::$key, handler) } }; } impl VoiceGateway { pub fn new(c: libremetaverse::GridClient) -> Result { let inner = Arc::new(GatewayInner { _client: c, pipe: TCPPipe::new()?, state: Mutex::new(GatewayState::default()), events: EventBus::default(), pipe_subscriptions: Mutex::new(Vec::new()), disposed: AtomicBool::new(false), }); let weak = Arc::downgrade(&inner); let received = inner .pipe .subscribe_on_receive_line(Some(TCPPipeOnReceiveLineCallback::from_handler( move |line| { weak.upgrade() .ok_or(Error::InvalidOperation) .and_then(|inner| inner.handle_line(&line)) }, ))); let weak = Arc::downgrade(&inner); let disconnected = inner.pipe.subscribe_on_disconnected(Some( TCPPipeOnDisconnectedCallback::from_handler(move |_| { let inner = weak.upgrade().ok_or(Error::InvalidOperation)?; let mut state = lock(&inner.state); state.daemon_connected = false; drop(state); inner.events.emit(EventKey::DaemonDisconnected, ()); Ok(()) }), )); *lock(&inner.pipe_subscriptions) = vec![received, disconnected]; Ok(Self { inner }) } /// Feeds one already-framed control line through the same bounded parser /// used by the TCP reader. This is shared by the legacy manager facade. #[cfg(test)] pub(crate) fn process_control_line(&self, line: &str) -> Result<(), Error> { self.inner.handle_line(line) } pub(crate) fn subscribe_control_line(&self, handler: EventHandler) -> Subscription { self.inner .events .subscribe(EventKey::RawControlLine, Some(handler)) } event_subscription!( subscribe_on_account_login_response, AccountLoginResponse, VoiceGatewayVoiceAccountEventArgs ); event_subscription!( subscribe_on_account_login_state_change_event, AccountLoginState, VoiceGatewayAccountLoginStateChangeEventArgs ); event_subscription!( subscribe_on_aux_audio_properties_event, AuxAudioProperties, VoiceGatewayAudioPropertiesEventArgs ); event_subscription!( subscribe_on_aux_get_capture_devices_response, CaptureDevices, VoiceGatewayVoiceDevicesEventArgs ); event_subscription!( subscribe_on_aux_get_render_devices_response, RenderDevices, VoiceGatewayVoiceDevicesEventArgs ); event_subscription!( subscribe_on_connector_create_response, ConnectorCreate, VoiceGatewayVoiceConnectorEventArgs ); zero_callback_subscription!( subscribe_on_daemon_connected, DaemonConnected, VoiceGatewayDaemonConnectedCallback ); zero_callback_subscription!( subscribe_on_daemon_couldnt_connect, DaemonCouldntConnect, VoiceGatewayDaemonCouldntConnectCallback ); zero_callback_subscription!( subscribe_on_daemon_couldnt_run, DaemonCouldntRun, VoiceGatewayDaemonCouldntRunCallback ); zero_callback_subscription!( subscribe_on_daemon_disconnected, DaemonDisconnected, VoiceGatewayDaemonDisconnectedCallback ); zero_callback_subscription!( subscribe_on_daemon_exited, DaemonExited, VoiceGatewayDaemonExitedCallback ); zero_callback_subscription!( subscribe_on_daemon_running, DaemonRunning, VoiceGatewayDaemonRunningCallback ); event_subscription!( subscribe_on_session_added_event, SessionAdded, VoiceGatewaySessionAddedEventArgs ); event_subscription!(subscribe_on_session_create, SessionCreate, ()); event_subscription!( subscribe_on_session_create_response, SessionCreateResponse, VoiceGatewayVoiceSessionEventArgs ); event_subscription!( subscribe_on_session_group_added_event, SessionGroupAdded, VoiceGatewaySessionGroupAddedEventArgs ); event_subscription!( subscribe_on_session_media_event, SessionMedia, VoiceGatewaySessionMediaEventArgs ); event_subscription!( subscribe_on_session_new_event, SessionNew, VoiceGatewayNewSessionEventArgs ); event_subscription!( subscribe_on_session_participant_added_event, ParticipantAdded, VoiceGatewayParticipantAddedEventArgs ); event_subscription!( subscribe_on_session_participant_properties_event, ParticipantProperties, VoiceGatewayParticipantPropertiesEventArgs ); event_subscription!( subscribe_on_session_participant_removed_event, ParticipantRemoved, VoiceGatewayParticipantRemovedEventArgs ); event_subscription!( subscribe_on_session_participant_state_change_event, ParticipantState, VoiceGatewayParticipantStateChangeEventArgs ); event_subscription!( subscribe_on_session_participant_updated_event, ParticipantUpdated, VoiceGatewayParticipantUpdatedEventArgs ); event_subscription!(subscribe_on_session_remove, SessionRemove, ()); event_subscription!( subscribe_on_session_removed_event, SessionRemoved, VoiceGatewaySessionRemovedEventArgs ); event_subscription!( subscribe_on_session_state_change_event, SessionState, VoiceGatewaySessionStateChangeEventArgs ); event_subscription!( subscribe_on_session_updated_event, SessionUpdated, VoiceGatewaySessionUpdatedEventArgs ); event_subscription!( subscribe_on_voice_response, VoiceResponse, VoiceGatewayVoiceResponseEventArgs ); #[must_use] pub fn subscribe_on_voice_connection_change( &self, handler: Option, ) -> Subscription { let handler = handler.map(|handler| { Arc::new(move |state| { let _ = handler.invoke(state); }) as EventHandler }); self.inner .events .subscribe(EventKey::VoiceConnection, handler) } #[must_use] pub fn subscribe_on_voice_mic_test( &self, handler: Option, ) -> Subscription { let handler = handler.map(|handler| { Arc::new(move |level| { let _ = handler.invoke(level); }) as EventHandler }); self.inner.events.subscribe(EventKey::VoiceMicTest, handler) } pub fn connect_to_daemon(&self, address: String, port: i32) -> Result { self.ensure_active_object()?; let connected = self.inner.pipe.connect(address, port)?.is_none(); { let mut state = lock(&self.inner.state); state.daemon_connected = connected; state.daemon_running = connected; } if connected { self.inner.events.emit(EventKey::DaemonRunning, ()); self.inner.events.emit(EventKey::DaemonConnected, ()); self.inner.events.emit( EventKey::VoiceConnection, VoiceGatewayConnectionState::DaemonConnected, ); } else { self.inner.events.emit(EventKey::DaemonCouldntConnect, ()); } Ok(connected) } pub fn start(&self) -> Result<(), Error> { self.ensure_active_object()?; lock(&self.inner.state).active = true; Ok(()) } pub fn stop(&self) -> Result<(), Error> { if self.inner.disposed.load(Ordering::Acquire) { return Ok(()); } let (session, account, connector, connected) = { let state = lock(&self.inner.state); ( state.session_handle.clone(), state.account_handle.clone(), state.connector_handle.clone(), state.daemon_connected, ) }; if connected { if !session.is_empty() { let _ = self.session_terminate(session); } if !account.is_empty() { let _ = self.account_logout(account); } if !connector.is_empty() { let _ = self.connector_initiate_shutdown(connector); } } self.inner.pipe.disconnect()?; let mut state = lock(&self.inner.state); state.active = false; state.daemon_connected = false; state.daemon_running = false; state.connector_handle.clear(); state.account_handle.clear(); state.session_handle.clear(); Ok(()) } pub fn dispose(&self) -> Result<(), Error> { if self.inner.disposed.swap(true, Ordering::AcqRel) { return Ok(()); } self.stop_after_dispose() } fn stop_after_dispose(&self) -> Result<(), Error> { self.inner.pipe.disconnect()?; let mut state = lock(&self.inner.state); state.active = false; state.daemon_connected = false; state.daemon_running = false; Ok(()) } /// Native releases connect to an explicitly supplied service and never /// spawn a proprietary daemon or child process. pub fn start_daemon(&self, path: String, args: String) -> Result<(), Error> { if path.is_empty() || args.len() > 64 * 1024 { return Err(Error::Argument); } self.inner.events.emit(EventKey::DaemonCouldntRun, ()); Err(Error::InvalidOperation) } pub fn stop_daemon(&self) -> Result<(), Error> { let was_running = lock(&self.inner.state).daemon_running; self.inner.pipe.disconnect()?; let mut state = lock(&self.inner.state); state.daemon_running = false; state.daemon_connected = false; drop(state); if was_running { self.inner.events.emit(EventKey::DaemonExited, ()); } Ok(()) } pub fn request_with_string(&self, action: String) -> Result { self.request_with_string_string(action, None) } pub fn request_with_string_string( &self, action: String, request_xml: Option, ) -> Result { self.ensure_active_object()?; if !valid_xml_name(&action) { return Err(Error::Argument); } let connected = lock(&self.inner.state).daemon_connected; if !connected { return Ok(-1); } let body = request_xml.unwrap_or_default(); validate_fragment(&body)?; let request_id = { let mut state = lock(&self.inner.state); let id = state.request_id; state.request_id = state .request_id .checked_add(1) .ok_or(Error::InvalidOperation)?; id }; let request = if body.is_empty() { format!("\n\n\n") } else { format!( "{body}\n\n\n" ) }; if request.len() > MAX_XML_BYTES { return Err(Error::Argument); } let bytes = request .chars() .map(|character| { if character.is_ascii() { character as u8 } else { b'?' } }) .collect(); match self.inner.pipe.send_data(bytes) { Ok(()) => Ok(request_id), Err(Error::Socket | Error::InvalidOperation) => Ok(-1), Err(error) => Err(error), } } pub fn make_xml(name: String, text: String) -> Result { xml_element(&name, &text) } pub fn account_login( &self, connector_handle: String, account_name: String, account_password: String, audio_session_answer_mode: String, account_uri: String, participant_property_frequency: i32, enable_buddies_and_presence: bool, ) -> Result { let body = xml_fields(&[ ("ConnectorHandle", connector_handle), ("AccountName", account_name), ("AccountPassword", account_password), ("AudioSessionAnswerMode", audio_session_answer_mode), ("AccountURI", account_uri), ( "ParticipantPropertyFrequency", participant_property_frequency.to_string(), ), ( "EnableBuddiesAndPresence", enable_buddies_and_presence.to_string(), ), ("BuddyManagementMode", "Application".into()), ])?; self.request_with_string_string("Account.Login.1".into(), Some(body)) } pub fn account_logout(&self, account_handle: String) -> Result { self.request_field("Account.Logout.1", "AccountHandle", account_handle) } pub fn aux_get_capture_devices(&self) -> Result { self.request_with_string("Aux.GetCaptureDevices.1".into()) } pub fn aux_get_render_devices(&self) -> Result { self.request_with_string("Aux.GetRenderDevices.1".into()) } pub fn aux_set_render_device(&self, value: String) -> Result { self.request_field("Aux.SetRenderDevice.1", "RenderDeviceSpecifier", value) } pub fn aux_set_capture_device(&self, value: String) -> Result { self.request_field("Aux.SetCaptureDevice.1", "CaptureDeviceSpecifier", value) } pub fn aux_capture_audio_start(&self, duration: i32) -> Result { self.request_field("Aux.CaptureAudioStart.1", "Duration", duration.to_string()) } pub fn aux_capture_audio_stop(&self) -> Result { self.request_with_string("Aux.CaptureAudioStop.1".into()) } pub fn aux_set_mic_level(&self, level: i32) -> Result { self.request_field("Aux.SetMicLevel.1", "Level", level.to_string()) } pub fn aux_set_speaker_level(&self, level: i32) -> Result { self.request_field("Aux.SetSpeakerLevel.1", "Level", level.to_string()) } pub fn connector_create( &self, client_name: String, account_management_server: String, minimum_port: u16, maximum_port: u16, logging: VoiceGatewayVoiceLoggingSettings, ) -> Result { if minimum_port > maximum_port { return Err(Error::Argument); } let logging_xml = xml_fields(&[ ("Enabled", logging.enabled.to_string()), ("Folder", logging.folder), ("FileNamePrefix", logging.file_name_prefix), ("FileNameSuffix", logging.file_name_suffix), ("LogLevel", logging.log_level.to_string()), ])?; let body = format!( "{}{logging_xml}", xml_fields(&[ ("ClientName", client_name), ("AccountManagementServer", account_management_server), ("MinimumPort", minimum_port.to_string()), ("MaximumPort", maximum_port.to_string()), ("Mode", "Normal".into()), ])? ); self.request_with_string_string("Connector.Create.1".into(), Some(body)) } pub fn connector_initiate_shutdown(&self, handle: String) -> Result { self.request_field("Connector.InitiateShutdown.1", "ConnectorHandle", handle) } pub fn connector_mute_local_mic(&self, handle: String, mute: bool) -> Result { self.connector_value_request("Connector.MuteLocalMic.1", handle, mute.to_string()) } pub fn connector_mute_local_speaker(&self, handle: String, mute: bool) -> Result { self.connector_value_request("Connector.MuteLocalSpeaker.1", handle, mute.to_string()) } pub fn connector_set_local_mic_volume(&self, handle: String, value: i32) -> Result { self.connector_value_request("Connector.SetLocalMicVolume.1", handle, value.to_string()) } pub fn connector_set_local_speaker_volume( &self, handle: String, value: i32, ) -> Result { self.connector_value_request( "Connector.SetLocalSpeakerVolume.1", handle, value.to_string(), ) } pub fn session_connect( &self, session_handle: String, audio_media: String, ) -> Result { self.request_fields_action( "Session.Connect.1", &[ ("SessionHandle", session_handle), ("AudioMedia", audio_media), ], ) } #[allow(clippy::too_many_arguments)] pub fn session_create( &self, account_handle: String, session_uri: String, name: String, password: String, join_audio: bool, join_text: bool, password_hash_algorithm: String, ) -> Result { let mut fields = vec![ ("AccountHandle", account_handle), ("URI", session_uri), ("Name", name), ]; if !password.is_empty() { fields.push(("Password", password)); fields.push(("PasswordHashAlgorithm", password_hash_algorithm)); } fields.extend([ ("ConnectAudio", join_audio.to_string()), ("ConnectText", join_text.to_string()), ("JoinAudio", join_audio.to_string()), ("JoinText", join_text.to_string()), ("VoiceFontID", "0".into()), ]); self.request_fields_action("Session.Create.1", &fields) } pub fn session_render_audio_start( &self, sound_file_path: String, loop_: bool, ) -> Result { self.request_fields_action( "Session.RenderAudioStart.1", &[ ("SoundFilePath", sound_file_path), ("Loop", if loop_ { "1" } else { "0" }.into()), ], ) } pub fn session_render_audio_stop(&self, path: String) -> Result { self.request_field("Session.RenderAudioStop.1", "SoundFilePath", path) } pub fn session_terminate(&self, handle: String) -> Result { self.request_field("Session.Terminate.1", "SessionHandle", handle) } pub fn session_set_participant_volume_for_me( &self, session_handle: String, participant_uri: String, volume: i32, ) -> Result { self.request_fields_action( "Session.SetParticipantVolumeForMe.1", &[ ("SessionHandle", session_handle), ("ParticipantURI", participant_uri), ("Volume", volume.to_string()), ], ) } pub(crate) fn session_set_participant_mute_for_me( &self, session_handle: String, participant_uri: String, mute: bool, ) -> Result { self.request_fields_action( "Session.SetParticipantMuteForMe.1", &[ ("SessionHandle", session_handle), ("ParticipantURI", participant_uri), ("Mute", if mute { "1" } else { "0" }.into()), ], ) } pub fn session_set3_d_position( &self, session_handle: String, speaker_position: VoicePosition, listener_position: VoicePosition, ) -> Result { let body = format!( "{}{}{}", xml_element("SessionHandle", &session_handle)?, position_xml(&speaker_position)?, position_xml(&listener_position)?, ); self.request_with_string_string("Session.Set3DPosition.1".into(), Some(body)) } pub fn sip_from_uuid(&self, id: UUID) -> Result { if id == UUID::zero() { return Ok(format!("sip:@{}", lock(&self.inner.state).sip_server)); } let encoded = base64::engine::general_purpose::STANDARD .encode(id.get_bytes()?) .replace('+', "-") .replace('/', "_"); Ok(format!( "sip:x{encoded}@{}", lock(&self.inner.state).sip_server )) } #[must_use] pub fn current_capture_device(&self) -> String { lock(&self.inner.state).current_capture_device.clone() } pub fn set_current_capture_device(&mut self, value: String) { lock(&self.inner.state).current_capture_device = value.clone(); let _ = self.aux_set_capture_device(value); } #[must_use] pub fn playback_device(&self) -> String { lock(&self.inner.state).current_playback_device.clone() } pub fn set_playback_device(&mut self, value: String) { lock(&self.inner.state).current_playback_device = value.clone(); let _ = self.aux_set_render_device(value); } #[must_use] pub fn daemon_is_connected(&self) -> bool { lock(&self.inner.state).daemon_connected } pub fn set_daemon_is_connected(&mut self, value: bool) { lock(&self.inner.state).daemon_connected = value; } #[must_use] pub fn daemon_is_running(&self) -> bool { lock(&self.inner.state).daemon_running } pub fn set_daemon_is_running(&mut self, value: bool) { lock(&self.inner.state).daemon_running = value; } #[must_use] pub fn request_id(&self) -> i32 { lock(&self.inner.state).request_id } pub fn set_request_id(&mut self, value: i32) { lock(&self.inner.state).request_id = value; } pub fn set_mic_level(&mut self, value: i32) { let handle = lock(&self.inner.state).connector_handle.clone(); let _ = self.connector_set_local_mic_volume(handle, value); } pub fn set_spkr_level(&mut self, value: i32) { let handle = lock(&self.inner.state).connector_handle.clone(); let _ = self.connector_set_local_speaker_volume(handle, value); } pub fn set_mic_mute(&mut self, value: bool) { let handle = lock(&self.inner.state).connector_handle.clone(); let _ = self.connector_mute_local_mic(handle, value); } pub fn set_spkr_mute(&mut self, value: bool) { let handle = lock(&self.inner.state).connector_handle.clone(); let _ = self.connector_mute_local_speaker(handle, value); } #[must_use] pub fn test_mode(&self) -> bool { lock(&self.inner.state).testing } pub fn set_test_mode(&mut self, value: bool) { lock(&self.inner.state).testing = value; if value { let _ = self.aux_capture_audio_start(0); } else { let _ = self.aux_capture_audio_stop(); } } fn ensure_active_object(&self) -> Result<(), Error> { if self.inner.disposed.load(Ordering::Acquire) { Err(Error::InvalidOperation) } else { Ok(()) } } fn request_field(&self, action: &str, name: &str, value: String) -> Result { self.request_fields_action(action, &[(name, value)]) } fn request_fields_action(&self, action: &str, fields: &[(&str, String)]) -> Result { self.request_with_string_string(action.into(), Some(xml_fields(fields)?)) } fn connector_value_request( &self, action: &str, handle: String, value: String, ) -> Result { self.request_fields_action(action, &[("ConnectorHandle", handle), ("Value", value)]) } } impl GatewayInner { fn handle_line(self: &Arc, line: &str) -> Result<(), Error> { if line.len() > MAX_XML_BYTES || line.contains(" self.handle_response(root), "Event" => self.handle_event(root), _ => Err(Error::Argument), } } fn handle_response(self: &Arc, root: Node<'_, '_>) -> Result<(), Error> { let action = root.attribute("action").ok_or(Error::Argument)?; let return_code = field_i32(root, "ReturnCode"); let status_code = field_i32(root, "StatusCode"); let status = field(root, "StatusString"); match action { "Connector.Create.1" => { let version = field(root, "VersionID"); let handle = field(root, "ConnectorHandle"); lock(&self.state).connector_handle = handle.clone(); self.events.emit( EventKey::ConnectorCreate, VoiceGatewayVoiceConnectorEventArgs::new( return_code, status_code, status, version, handle, )?, ); } "Aux.GetCaptureDevices.1" | "Aux.GetRenderDevices.1" => { let capture = action.contains("Capture"); let item = if capture { "CaptureDevice" } else { "RenderDevice" }; let current_item = if capture { "CurrentCaptureDevice" } else { "CurrentRenderDevice" }; let devices = device_values(root, item); let current = nested_device(root, current_item); let type_ = if capture { VoiceGatewayResponseType::GetCaptureDevices } else { VoiceGatewayResponseType::GetRenderDevices }; let key = if capture { lock(&self.state).current_capture_device = current.clone(); EventKey::CaptureDevices } else { lock(&self.state).current_playback_device = current.clone(); EventKey::RenderDevices }; self.events.emit( key, VoiceGatewayVoiceDevicesEventArgs::new( type_, return_code, status_code, status, current, devices, )?, ); } "Account.Login.1" => { let handle = field(root, "AccountHandle"); lock(&self.state).account_handle = handle.clone(); self.events.emit( EventKey::AccountLoginResponse, VoiceGatewayVoiceAccountEventArgs::new( return_code, status_code, status, handle, )?, ); } "Session.Create.1" => { let handle = field(root, "SessionHandle"); let mut state = lock(&self.state); state.session_handle = handle.clone(); if !handle.is_empty() && return_code == 0 && status_code == 0 { let gateway = VoiceGateway { inner: Arc::clone(self), }; state .sessions .insert(handle.clone(), VoiceSession::new(gateway, handle.clone())?); } drop(state); self.events.emit( EventKey::SessionCreateResponse, VoiceGatewayVoiceSessionEventArgs::new( return_code, status_code, status, handle, )?, ); self.events.emit(EventKey::SessionCreate, ()); } _ => { if let Some(type_) = response_type(action) { self.events.emit( EventKey::VoiceResponse, VoiceGatewayVoiceResponseEventArgs::new( type_, return_code, status_code, status, )?, ); } } } Ok(()) } #[allow(clippy::too_many_lines)] fn handle_event(self: &Arc, root: Node<'_, '_>) -> Result<(), Error> { match root.attribute("type").ok_or(Error::Argument)? { "LoginStateChangeEvent" | "AccountLoginStateChangeEvent" => self.events.emit( EventKey::AccountLoginState, VoiceGatewayAccountLoginStateChangeEventArgs::new( field(root, "AccountHandle"), field_i32(root, "StatusCode"), field(root, "StatusString"), login_state(field_i32(root, "State")), )?, ), "SessionNewEvent" => self.events.emit( EventKey::SessionNew, VoiceGatewayNewSessionEventArgs::new( field(root, "AccountHandle"), field(root, "SessionHandle"), field(root, "URI"), field_bool(root, "IsChannel"), field(root, "Name"), field(root, "AudioMedia"), )?, ), "SessionStateChangeEvent" => self.events.emit( EventKey::SessionState, VoiceGatewaySessionStateChangeEventArgs::new( field(root, "SessionHandle"), field_i32(root, "StatusCode"), field(root, "StatusString"), session_state(field_i32(root, "State")), field(root, "URI"), field_bool(root, "IsChannel"), field(root, "ChannelName"), )?, ), "ParticipantAddedEvent" => { let session_handle = field(root, "SessionHandle"); let participant_uri = field(root, "ParticipantUri"); if let Some(session) = lock(&self.state).sessions.get(&session_handle).cloned() { session.add_participant(participant_uri.clone())?; } self.events.emit( EventKey::ParticipantAdded, VoiceGatewayParticipantAddedEventArgs::new( field(root, "SessionGroupHandle"), session_handle, participant_uri, field(root, "AccountName"), field(root, "DisplayName"), participant_type(field_i32(root, "ParticipantType")), field(root, "Application"), )?, ); } "ParticipantRemovedEvent" => { let session_handle = field(root, "SessionHandle"); let participant_uri = field(root, "ParticipantUri"); if let Some(session) = lock(&self.state).sessions.get(&session_handle).cloned() { session.remove_participant(&participant_uri); } self.events.emit( EventKey::ParticipantRemoved, VoiceGatewayParticipantRemovedEventArgs::new( field(root, "SessionGroupHandle"), session_handle, participant_uri, field(root, "AccountName"), field(root, "Reason"), )?, ); } "ParticipantStateChangeEvent" => self.events.emit( EventKey::ParticipantState, VoiceGatewayParticipantStateChangeEventArgs::new( field(root, "SessionHandle"), field_i32(root, "StatusCode"), field(root, "StatusString"), participant_state(field_i32(root, "State")), field(root, "ParticipantUri"), field(root, "AccountName"), field(root, "DisplayName"), participant_type(field_i32(root, "ParticipantType")), )?, ), "ParticipantPropertiesEvent" => { let session_handle = field(root, "SessionHandle"); let participant_uri = field(root, "ParticipantUri"); let locally_muted = field_bool(root, "IsLocallyMuted"); let moderator_muted = field_bool(root, "IsModeratorMuted"); let speaking = field_bool(root, "IsSpeaking"); let volume = field_i32(root, "Volume"); let energy = field_f32(root, "Energy"); if let Some(session) = lock(&self.state).sessions.get(&session_handle).cloned() { session.update_participant( &participant_uri, locally_muted || moderator_muted, speaking, volume, energy, ); } self.events.emit( EventKey::ParticipantProperties, VoiceGatewayParticipantPropertiesEventArgs::new( session_handle, participant_uri, locally_muted, moderator_muted, speaking, volume, energy, )?, ); } "ParticipantUpdatedEvent" => { let session_handle = field(root, "SessionHandle"); let participant_uri = field(root, "ParticipantUri"); let moderator_muted = field_bool(root, "IsModeratorMuted"); let speaking = field_bool(root, "IsSpeaking"); let volume = field_i32(root, "Volume"); let energy = field_f32(root, "Energy"); if let Some(session) = lock(&self.state).sessions.get(&session_handle).cloned() { session.update_participant( &participant_uri, moderator_muted, speaking, volume, energy, ); } self.events.emit( EventKey::ParticipantUpdated, VoiceGatewayParticipantUpdatedEventArgs::new( session_handle, participant_uri, moderator_muted, speaking, volume, energy, )?, ); } "SessionGroupAddedEvent" => self.events.emit( EventKey::SessionGroupAdded, VoiceGatewaySessionGroupAddedEventArgs::new( field(root, "AccountHandle"), field(root, "SessionGroupHandle"), field(root, "Type"), )?, ), "SessionAddedEvent" => self.events.emit( EventKey::SessionAdded, VoiceGatewaySessionAddedEventArgs::new( field(root, "SessionGroupHandle"), field(root, "SessionHandle"), field(root, "Uri"), field_bool(root, "IsChannel"), field_bool(root, "Incoming"), )?, ), "SessionRemovedEvent" => { let session_handle = field(root, "SessionHandle"); if let Some(session) = lock(&self.state).sessions.remove(&session_handle) { session.close(); } self.events.emit( EventKey::SessionRemoved, VoiceGatewaySessionRemovedEventArgs::new( field(root, "SessionGroupHandle"), session_handle, field(root, "Uri"), )?, ); self.events.emit(EventKey::SessionRemove, ()); } "SessionUpdatedEvent" => self.events.emit( EventKey::SessionUpdated, VoiceGatewaySessionUpdatedEventArgs::new( field(root, "SessionGroupHandle"), field(root, "SessionHandle"), field(root, "Uri"), field_i32(root, "IsMuted") != 0, field_i32(root, "Volume"), field_i32(root, "TransmitEnabled") != 0, field_i32(root, "IsFocused") != 0, )?, ), "AuxAudioPropertiesEvent" => { let energy = field_f32(root, "MicEnergy"); self.events.emit( EventKey::AuxAudioProperties, VoiceGatewayAudioPropertiesEventArgs::new( field_bool(root, "MicIsActive"), energy, field_i32(root, "MicVolume"), field_i32(root, "SpeakerVolume"), )?, ); self.events.emit(EventKey::VoiceMicTest, energy); } "SessionMediaEvent" => self.events.emit( EventKey::SessionMedia, VoiceGatewaySessionMediaEventArgs::new( field(root, "SessionHandle"), field_bool(root, "HasText"), field_bool(root, "HasAudio"), field_bool(root, "HasVideo"), field_bool(root, "Terminated"), )?, ), _ => {} } Ok(()) } } fn valid_xml_name(value: &str) -> bool { !value.is_empty() && value.len() <= 128 && value .bytes() .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')) } fn validate_fragment(fragment: &str) -> Result<(), Error> { if fragment.len() > MAX_XML_BYTES || fragment.contains("{fragment}")).map_err(|_| Error::Argument)?; } Ok(()) } fn xml_element(name: &str, text: &str) -> Result { if !valid_xml_name(name) || text.len() > 64 * 1024 { return Err(Error::Argument); } if text.is_empty() { return Ok(format!("<{name} />")); } let mut escaped = String::with_capacity(text.len()); for character in text.chars() { match character { '&' => escaped.push_str("&"), '<' => escaped.push_str("<"), '>' => escaped.push_str(">"), '\'' => escaped.push_str("'"), '"' => escaped.push_str("""), '\0'..='\x08' | '\x0b' | '\x0c' | '\x0e'..='\x1f' => { return Err(Error::Argument); } _ => escaped.push(character), } } Ok(format!("<{name}>{escaped}")) } fn xml_fields(fields: &[(&str, String)]) -> Result { let mut output = String::new(); for (name, value) in fields { output.push_str(&xml_element(name, value)?); } Ok(output) } fn position_xml(position: &VoicePosition) -> Result { fn vector(name: &str, value: Vector3d) -> Result { Ok(format!( "<{name}>{}", xml_fields(&[ ("X", value.x.to_string()), ("Y", value.y.to_string()), ("Z", value.z.to_string()), ])? )) } Ok([ vector("Position", position.position)?, vector("Velocity", position.velocity)?, vector("AtOrientation", position.at_orientation)?, vector("UpOrientation", position.up_orientation)?, vector("LeftOrientation", position.left_orientation)?, ] .concat()) } fn field(root: Node<'_, '_>, name: &str) -> String { root.descendants() .find(|node| node.has_tag_name(name)) .and_then(|node| node.text()) .unwrap_or_default() .to_owned() } fn field_i32(root: Node<'_, '_>, name: &str) -> i32 { field(root, name).parse().unwrap_or_default() } fn field_f32(root: Node<'_, '_>, name: &str) -> f32 { field(root, name).parse().unwrap_or_default() } fn field_bool(root: Node<'_, '_>, name: &str) -> bool { matches!( field(root, name).to_ascii_lowercase().as_str(), "true" | "1" ) } fn nested_device(root: Node<'_, '_>, name: &str) -> String { root.descendants() .find(|node| node.has_tag_name(name)) .map(|node| field(node, "Device")) .unwrap_or_default() } fn device_values(root: Node<'_, '_>, name: &str) -> Vec { root.descendants() .filter(|node| node.has_tag_name(name)) .map(|node| field(node, "Device")) .filter(|value| !value.is_empty()) .collect() } fn response_type(action: &str) -> Option { Some(match action { "Connector.InitiateShutdown.1" => VoiceGatewayResponseType::ConnectorInitiateShutdown, "Aux.SetRenderDevice.1" => VoiceGatewayResponseType::SetRenderDevice, "Connector.MuteLocalMic.1" => VoiceGatewayResponseType::MuteLocalMic, "Connector.MuteLocalSpeaker.1" => VoiceGatewayResponseType::MuteLocalSpeaker, "Connector.SetLocalMicVolume.1" => VoiceGatewayResponseType::SetLocalMicVolume, "Connector.SetLocalSpeakerVolume.1" => VoiceGatewayResponseType::SetLocalSpeakerVolume, "Aux.SetCaptureDevice.1" => VoiceGatewayResponseType::SetCaptureDevice, "Session.RenderAudioStart.1" => VoiceGatewayResponseType::RenderAudioStart, "Session.RenderAudioStop.1" => VoiceGatewayResponseType::RenderAudioStop, "Aux.CaptureAudioStart.1" => VoiceGatewayResponseType::CaptureAudioStart, "Aux.CaptureAudioStop.1" => VoiceGatewayResponseType::CaptureAudioStop, "Aux.SetMicLevel.1" => VoiceGatewayResponseType::SetMicLevel, "Aux.SetSpeakerLevel.1" => VoiceGatewayResponseType::SetSpeakerLevel, "Account.Logout.1" => VoiceGatewayResponseType::AccountLogout, "Session.Connect.1" => VoiceGatewayResponseType::SessionConnect, "Session.Terminate.1" => VoiceGatewayResponseType::SessionTerminate, "Session.SetParticipantVolumeForMe.1" => { VoiceGatewayResponseType::SetParticipantVolumeForMe } "Session.SetParticipantMuteForMe.1" => VoiceGatewayResponseType::SetParticipantMuteForMe, "Session.Set3DPosition.1" => VoiceGatewayResponseType::Set3DPosition, _ => return None, }) } fn login_state(value: i32) -> VoiceGatewayLoginState { match value { 1 => VoiceGatewayLoginState::LoggedIn, 4 => VoiceGatewayLoginState::Error, _ => VoiceGatewayLoginState::LoggedOut, } } fn session_state(value: i32) -> VoiceGatewaySessionState { match value { 2 => VoiceGatewaySessionState::Answering, 3 => VoiceGatewaySessionState::InProgress, 4 => VoiceGatewaySessionState::Connected, 5 => VoiceGatewaySessionState::Disconnected, 6 => VoiceGatewaySessionState::Hold, 7 => VoiceGatewaySessionState::Refer, 8 => VoiceGatewaySessionState::Ringing, _ => VoiceGatewaySessionState::Idle, } } fn participant_state(value: i32) -> VoiceGatewayParticipantState { match value { 2 => VoiceGatewayParticipantState::Pending, 3 => VoiceGatewayParticipantState::Incoming, 4 => VoiceGatewayParticipantState::Answering, 5 => VoiceGatewayParticipantState::InProgress, 6 => VoiceGatewayParticipantState::Ringing, 7 => VoiceGatewayParticipantState::Connected, 8 => VoiceGatewayParticipantState::Disconnecting, 9 => VoiceGatewayParticipantState::Disconnected, _ => VoiceGatewayParticipantState::Idle, } } fn participant_type(value: i32) -> VoiceGatewayParticipantType { match value { 1 => VoiceGatewayParticipantType::Moderator, 2 => VoiceGatewayParticipantType::Focus, _ => VoiceGatewayParticipantType::User, } } #[cfg(test)] mod tests { use super::*; use std::sync::mpsc; #[test] fn xml_builder_escapes_values_and_rejects_active_xml() { assert_eq!( VoiceGateway::make_xml("Name".into(), "<&>\"'".into()).unwrap(), "<&>"'" ); assert!(validate_fragment("").is_err()); assert!(VoiceGateway::make_xml("bad name".into(), "x".into()).is_err()); } #[test] fn response_and_event_dispatch_retain_typed_payloads() { let client = libremetaverse::GridClient::new().unwrap(); let gateway = VoiceGateway::new(client).unwrap(); let (sender, receiver) = mpsc::channel(); let _subscription = gateway.subscribe_on_session_participant_updated_event(Some(Arc::new(move |event| { sender.send(event).unwrap(); }))); gateway.inner.handle_line("ssip:p@example.testtrue1420.75").unwrap(); let event = receiver.recv().unwrap(); assert_eq!(event.session_handle, "s"); assert!(event.is_muted); assert!(event.is_speaking); assert_eq!(event.volume, 42); assert_eq!(event.energy, 0.75); } #[test] fn protocol_events_drive_the_session_participant_lifecycle() { let gateway = VoiceGateway::new(libremetaverse::GridClient::new().unwrap()).unwrap(); gateway.inner.handle_line("00OKsession-3").unwrap(); let session = lock(&gateway.inner.state) .sessions .get("session-3") .cloned() .unwrap(); let (sender, receiver) = mpsc::channel(); let added = sender.clone(); let _added = session.subscribe_on_participant_added(Some(Arc::new(move |()| { added.send("added").unwrap(); }))); let updated = sender.clone(); let _updated = session.subscribe_on_participant_update(Some(Arc::new(move |()| { updated.send("updated").unwrap(); }))); let _removed = session.subscribe_on_participant_removed(Some(Arc::new(move |()| { sender.send("removed").unwrap(); }))); gateway.inner.handle_line("session-3sip:p@example.test").unwrap(); gateway.inner.handle_line("session-3sip:p@example.test1420.75").unwrap(); gateway.inner.handle_line("session-3sip:p@example.test").unwrap(); assert_eq!(receiver.recv().unwrap(), "added"); assert_eq!(receiver.recv().unwrap(), "updated"); assert_eq!(receiver.recv().unwrap(), "removed"); gateway.inner.handle_line("session-3").unwrap(); assert!( !lock(&gateway.inner.state) .sessions .contains_key("session-3") ); } }