diff --git a/core/main/src/broker/endpoint_broker.rs b/core/main/src/broker/endpoint_broker.rs index ae2c85d16..0fd0c28a9 100644 --- a/core/main/src/broker/endpoint_broker.rs +++ b/core/main/src/broker/endpoint_broker.rs @@ -81,19 +81,47 @@ pub struct BrokerSender { pub sender: Sender, } +/// Enum to distinguish between connection-level and session-level cleanup +#[derive(Clone, Debug)] +pub enum CleanupType { + /// Cleanup by connection_id - used when a single WebSocket connection disconnects + Connection(String), + /// Cleanup by session_id - used when an app session is unloaded/terminated + Session(String), +} + #[derive(Clone, Debug, Default)] pub struct BrokerCleaner { - pub cleaner: Option>, + pub cleaner: Option>, } impl BrokerCleaner { - async fn cleanup_session(&self, appid: &str) -> Result { + /// Cleanup subscriptions for a specific connection (WebSocket disconnect) + pub async fn cleanup_connection(&self, connection_id: &str) -> Result { if let Some(cleaner) = self.cleaner.clone() { - if let Err(e) = cleaner.send(appid.to_owned()).await { - error!("Could not clean up {} {:?}", appid, e); + if let Err(e) = cleaner + .send(CleanupType::Connection(connection_id.to_owned())) + .await + { + error!("Could not clean up connection {} {:?}", connection_id, e); return Err(RippleError::SendFailure); } - return Ok(appid.to_owned()); + return Ok(connection_id.to_owned()); + } + Err(RippleError::NotAvailable) + } + + /// Cleanup all subscriptions for a session (app unload/termination) + pub async fn cleanup_session(&self, session_id: &str) -> Result { + if let Some(cleaner) = self.cleaner.clone() { + if let Err(e) = cleaner + .send(CleanupType::Session(session_id.to_owned())) + .await + { + error!("Could not clean up session {} {:?}", session_id, e); + return Err(RippleError::SendFailure); + } + return Ok(session_id.to_owned()); } Err(RippleError::NotAvailable) } @@ -1042,7 +1070,7 @@ impl EndpointBrokerState { Ok(response) } RenderedRequest::BrokerRequest(request) => { - info!( + trace!( "Sending broker_request json rpc response to endpoint {:?}", request ); @@ -1094,20 +1122,31 @@ impl EndpointBrokerState { } } - // Method to cleanup all subscription on App termination - pub async fn cleanup_for_app(&self, id: &str) { + /// Cleanup subscriptions for a specific connection (WebSocket disconnect). + /// This only removes subscriptions made by this specific connection, + /// not all subscriptions for the session. + pub async fn cleanup_for_connection(&self, connection_id: &str) { let cleaners = { self.cleaner_list.read().unwrap().clone() }; for cleaner in cleaners { - /* - for now, just eat the error - the return type was mainly added to prepate for future refactoring/testability - */ - let _ = cleaner.cleanup_session(id).await; + let _ = cleaner.cleanup_connection(connection_id).await; } // Clean up subscription entries from request_map and extension_request_map - // that belong to this app. These are never removed on disconnect otherwise. - self.cleanup_request_maps(id); + self.cleanup_request_maps(connection_id); + } + + /// Cleanup all subscriptions for a session (app unload/termination). + /// This removes ALL subscriptions associated with this session_id. + pub async fn cleanup_for_session(&self, session_id: &str) { + let cleaners = { self.cleaner_list.read().unwrap().clone() }; + + for cleaner in cleaners { + let _ = cleaner.cleanup_session(session_id).await; + } + + // Clean up subscription entries from request_map and extension_request_map + self.cleanup_request_maps(session_id); } /// Remove subscription/event entries from request_map and extension_request_map @@ -1137,7 +1176,7 @@ impl EndpointBrokerState { for id in &removed_ids { extn_map.remove(id); } - debug!( + trace!( "cleanup_request_maps: removed {} request_map and extension_request_map entries", removed_ids.len() ); @@ -1489,7 +1528,7 @@ impl BrokerOutputForwarder { .emit_debug(); let session_id = rpc_request.ctx.get_id(); if let Some(workflow_callback) = workflow_callback { - debug!("sending to workflow callback {:?}", response); + trace!("sending to workflow callback {:?}", response); LogSignal::new( "forward_response".to_string(), "sending to workflow callback".to_string(), @@ -3454,7 +3493,7 @@ mod endpoint_broker_tests { mod cleaner { use ripple_sdk::tokio::{self, sync::mpsc}; - use crate::broker::endpoint_broker::BrokerCleaner; + use crate::broker::endpoint_broker::{BrokerCleaner, CleanupType}; #[tokio::test] async fn test_cleanup_session_with_cleaner() { @@ -3463,7 +3502,23 @@ mod endpoint_broker_tests { assert!(cleaner.cleanup_session("test_app").await.is_ok()); let received = rx.recv().await; - assert_eq!(received, Some("test_app".to_string())); + assert!(matches!( + received, + Some(CleanupType::Session(ref s)) if s == "test_app" + )); + } + + #[tokio::test] + async fn test_cleanup_connection_with_cleaner() { + let (tx, mut rx) = mpsc::channel(1); + let cleaner = BrokerCleaner { cleaner: Some(tx) }; + + assert!(cleaner.cleanup_connection("test_cid").await.is_ok()); + let received = rx.recv().await; + assert!(matches!( + received, + Some(CleanupType::Connection(ref c)) if c == "test_cid" + )); } #[tokio::test] @@ -3473,6 +3528,14 @@ mod endpoint_broker_tests { // Should not panic or send anything assert!(cleaner.cleanup_session("test_app").await.is_err()); } + + #[tokio::test] + async fn test_cleanup_connection_without_cleaner() { + let cleaner = BrokerCleaner { cleaner: None }; + + // Should not panic or send anything + assert!(cleaner.cleanup_connection("test_cid").await.is_err()); + } } #[cfg(test)] mod workflow { diff --git a/core/main/src/broker/rules/rules_engine.rs b/core/main/src/broker/rules/rules_engine.rs index 24be1a24c..e0102c666 100644 --- a/core/main/src/broker/rules/rules_engine.rs +++ b/core/main/src/broker/rules/rules_engine.rs @@ -539,9 +539,11 @@ fn get_parse_ctx() -> MutexGuard<'static, ParseCtx> { } pub fn jq_compile(input: Value, filter: &str, reference: String) -> Result { - info!( + trace!( "Jq rule {} input {:?}, reference {}", - filter, input, reference + filter, + input, + reference ); let start = Utc::now().timestamp_millis(); // start out only from core filters, diff --git a/core/main/src/broker/service_broker.rs b/core/main/src/broker/service_broker.rs index 50a217336..006eaaab4 100644 --- a/core/main/src/broker/service_broker.rs +++ b/core/main/src/broker/service_broker.rs @@ -19,9 +19,10 @@ use super::endpoint_broker::{ EndpointBroker, EndpointBrokerState, BROKER_CHANNEL_BUFFER_SIZE, }; use crate::state::platform_state::PlatformState; +use jsonrpsee::tracing::trace; use ripple_sdk::{ api::{gateway::rpc_gateway_api::JsonRpcApiError, observability::log_signal::LogSignal}, - log::{error, info}, + log::error, service::service_message::{Id, ServiceMessage}, tokio::{self, sync::mpsc}, tokio_tungstenite::tungstenite::Message, @@ -121,7 +122,7 @@ impl ServiceBroker { } let message = Message::Text(request.clone()); - info!("Sending request to service {}: {:#?}", service_id, message); + trace!("Sending request to service {}: {:#?}", service_id, message); if let Err(err) = service_sender.try_send(message) { error!( @@ -171,7 +172,7 @@ impl ServiceBroker { fn update_service_request(broker_request: &BrokerRequest) -> Result { let v = Self::apply_request_rule(broker_request)?; - info!("transformed request {:?}", v); + trace!("transformed request {:?}", v); // Create a ServiceMessage let mut request = ServiceMessage::new_request( diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index 34903c17d..37dd83a75 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -17,7 +17,7 @@ use super::{ endpoint_broker::{ BrokerCallback, BrokerCleaner, BrokerConnectRequest, BrokerOutput, BrokerRequest, - BrokerSender, BrokerSubMap, EndpointBroker, EndpointBrokerState, + BrokerSender, BrokerSubMap, CleanupType, EndpointBroker, EndpointBrokerState, BROKER_CHANNEL_BUFFER_SIZE, }, thunder::thunder_plugins_status_mgr::StatusManager, @@ -198,7 +198,7 @@ impl ThunderBroker { ) -> Self { let endpoint = request.endpoint.clone(); let (broker_request_tx, mut broker_request_rx) = mpsc::channel(BROKER_CHANNEL_BUFFER_SIZE); - let (c_tx, mut c_tr) = mpsc::channel(2); + let (c_tx, mut c_tr) = mpsc::channel::(2); let broker_sender = BrokerSender { sender: broker_request_tx, }; @@ -291,7 +291,7 @@ impl ThunderBroker { }, Some(mut request) = broker_request_rx.recv() => { - debug!("Got request from receiver for broker {:?}", request); + trace!("Got request from receiver for broker {:?}", request); diagnostic_context.lock().await.replace(request.clone()); match broker_c.check_and_generate_plugin_activation_request(&request) { @@ -369,13 +369,62 @@ impl ThunderBroker { }, Some(cleanup_request) = c_tr.recv() => { - let value = { - broker_for_cleanup.subscription_map.write().unwrap().remove(&cleanup_request) + // Collect subscriptions to clean up based on cleanup type + let subscriptions_to_cleanup: Vec = match &cleanup_request { + CleanupType::Connection(connection_id) => { + // Connection cleanup: remove only subscriptions with matching cid + let mut sub_map = broker_for_cleanup.subscription_map.write().unwrap(); + let mut to_cleanup = Vec::new(); + + // Iterate all entries and remove matching subscriptions + let keys: Vec = sub_map.keys().cloned().collect(); + for key in keys { + if let Some(mut requests) = sub_map.remove(&key) { + let (matching, remaining): (Vec<_>, Vec<_>) = requests + .drain(..) + .partition(|req| req.rpc.ctx.cid.as_deref() == Some(connection_id)); + to_cleanup.extend(matching); + if !remaining.is_empty() { + sub_map.insert(key, remaining); + } + } + } + debug!( + "BrokerCleaner: connection cleanup for cid={}, found {} subscription(s)", + connection_id, to_cleanup.len() + ); + to_cleanup + } + CleanupType::Session(session_id) => { + // Session cleanup: remove all subscriptions for this session_id + let mut sub_map = broker_for_cleanup.subscription_map.write().unwrap(); + let mut to_cleanup = Vec::new(); + + // Iterate all entries and remove matching subscriptions + let keys: Vec = sub_map.keys().cloned().collect(); + for key in keys { + if let Some(mut requests) = sub_map.remove(&key) { + let (matching, remaining): (Vec<_>, Vec<_>) = requests + .drain(..) + .partition(|req| req.rpc.ctx.session_id == *session_id); + to_cleanup.extend(matching); + if !remaining.is_empty() { + sub_map.insert(key, remaining); + } + } + } + debug!( + "BrokerCleaner: session cleanup for session_id={}, found {} subscription(s)", + session_id, to_cleanup.len() + ); + to_cleanup + } }; - if let Some(mut cleanup) = value { - debug!( + + if !subscriptions_to_cleanup.is_empty() { + trace!( "BrokerCleaner: unsubscribing {} subscription(s)", - cleanup.len() + subscriptions_to_cleanup.len() ); // Send unregister calls directly to Thunder via the WebSocket. // We must NOT route through the broker sender + prepare_request(), @@ -383,7 +432,7 @@ impl ThunderBroker { // subscription_map — but we already removed entries above. let binding = ws_tx_wrap.clone(); let mut ws_tx = binding.lock().await; - while let Some(v) = cleanup.pop() { + for v in subscriptions_to_cleanup { let (callsign, method) = ThunderBroker::get_callsign_and_method_from_alias(&v.rule.alias); if let Some(method) = method { @@ -395,7 +444,7 @@ impl ThunderBroker { "id": format!("{}", v.rpc.ctx.call_id) } }); - debug!( + trace!( "BrokerCleaner: sending Thunder unregister for {}.{}", callsign, method ); @@ -419,8 +468,8 @@ impl ThunderBroker { ); } } else { - debug!( - "BrokerCleaner: no subscriptions found, skipping" + trace!( + "BrokerCleaner: no subscriptions found for cleanup, skipping" ); } } @@ -500,56 +549,71 @@ impl ThunderBroker { let callsign = collection.join("."); (callsign, method) } + + /// Get the subscription map key from the request context. + /// Uses connection_id (cid) if available, falls back to session_id. + fn get_subscription_key(request: &BrokerRequest) -> String { + request + .rpc + .ctx + .cid + .clone() + .unwrap_or_else(|| request.rpc.ctx.session_id.clone()) + } + fn unsubscribe(&self, request: &BrokerRequest) -> Option { let mut sub_map = self.subscription_map.write().unwrap(); + let sub_key = Self::get_subscription_key(request); trace!( - "Unsubscribing a listen request for session id: {:?}", - request.rpc.ctx.session_id + "Unsubscribing a listen request for key: {:?} (session_id: {:?}, cid: {:?})", + sub_key, + request.rpc.ctx.session_id, + request.rpc.ctx.cid ); - let app_id = &request.rpc.ctx.session_id; let method = &request.rpc.ctx.method; let mut existing_request = None; - if let Some(mut existing_requests) = sub_map.remove(app_id) { + if let Some(mut existing_requests) = sub_map.remove(&sub_key) { if let Some(i) = existing_requests .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) { existing_request = Some(existing_requests.remove(i)); } - let _ = sub_map.insert(app_id.clone(), existing_requests); + if !existing_requests.is_empty() { + let _ = sub_map.insert(sub_key.clone(), existing_requests); + } } existing_request } fn subscribe(&self, request: &BrokerRequest) -> Option { let mut sub_map = self.subscription_map.write().unwrap(); - let app_id = &request.rpc.ctx.session_id; + let sub_key = Self::get_subscription_key(request); let method = &request.rpc.ctx.method; let listen = request.rpc.is_listening(); let mut response = None; - debug!( - "Initial subscription map of {:?} app_id {:?}", - sub_map, app_id - ); - if let Some(mut v) = sub_map.remove(app_id) { - debug!("Subscription map after removing app {:?}", v); + if let Some(mut v) = sub_map.remove(&sub_key) { if let Some(i) = v .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) { - debug!( - "Removing subscription for method {} for app {}", - method, app_id + trace!( + "Removing subscription for method {} for key {}", + method, + sub_key ); response = Some(v.remove(i)); } if listen { v.push(request.clone()); } - let _ = sub_map.insert(app_id.clone(), v); - } else { - let _ = sub_map.insert(app_id.clone(), vec![request.clone()]); + // Only re-insert if the Vec is non-empty to avoid map bloat from empty entries + if !v.is_empty() { + let _ = sub_map.insert(sub_key.clone(), v); + } + } else if listen { + let _ = sub_map.insert(sub_key.clone(), vec![request.clone()]); } response } @@ -637,7 +701,7 @@ impl EndpointBroker for ThunderBroker { let mut requests = Vec::new(); let method = method.unwrap(); - debug!( + trace!( "Preparing request for method {} and callsign {} for {} subscription {}", method, callsign, @@ -650,6 +714,7 @@ impl EndpointBroker for ThunderBroker { let listen = rpc_request.rpc.is_listening(); // If there was an existing app and method combo for the same subscription just unregister that if let Some(cleanup) = self.subscribe(rpc_request) { + trace!("Sending Thunder unregister for {}.{}", callsign, method); requests.push( json!({ "jsonrpc": "2.0", @@ -666,6 +731,7 @@ impl EndpointBroker for ThunderBroker { // Given unregistration is already performed by previous step just do registration if listen { + trace!("Sending Thunder register for {}.{}", callsign, method); requests.push( json!({ "jsonrpc": "2.0", @@ -681,11 +747,7 @@ impl EndpointBroker for ThunderBroker { } } else if rpc_request.rpc.is_unlisten() { if let Some(cleanup) = self.unsubscribe(rpc_request) { - trace!( - "Unregistering thunder listener for call_id {} and method {}", - cleanup.rpc.ctx.call_id, - method - ); + trace!("Sending Thunder unregister for {}.{}", callsign, method); requests.push( json!({ "jsonrpc": "2.0", @@ -1380,8 +1442,8 @@ mod tests { thunder_broker.unsubscribe(&unsubscribe_request); let subscription_map = thunder_broker.subscription_map.write().unwrap(); - // TBD : Check. Unsubscribe is re-inserting the subscription back to the map - // let _ = sub_map.insert(app_id.clone(), existing_requests); - assert_eq!(subscription_map.len(), 1); + // After unsubscribing the only subscription for this key, the map entry should be removed + // (we no longer keep empty Vec entries in the map) + assert_eq!(subscription_map.len(), 0); } } diff --git a/core/main/src/broker/websocket_broker.rs b/core/main/src/broker/websocket_broker.rs index 15fadf3e5..72413082e 100644 --- a/core/main/src/broker/websocket_broker.rs +++ b/core/main/src/broker/websocket_broker.rs @@ -17,7 +17,7 @@ use super::endpoint_broker::{ BrokerCallback, BrokerCleaner, BrokerConnectRequest, BrokerOutputForwarder, BrokerRequest, - BrokerSender, EndpointBroker, + BrokerSender, CleanupType, EndpointBroker, }; use crate::broker::endpoint_broker::EndpointBrokerState; use crate::state::platform_state::PlatformState; @@ -42,7 +42,7 @@ impl WebsocketBroker { fn start(request: BrokerConnectRequest, callback: BrokerCallback) -> Self { let endpoint = request.endpoint.clone(); let (tx, mut tr) = mpsc::channel(10); - let (cleaner_tx, mut cleaner_tr) = mpsc::channel::(1); + let (cleaner_tx, mut cleaner_tr) = mpsc::channel::(1); let non_json_rpc_map: Arc>>>> = Arc::new(RwLock::new(HashMap::new())); let map_clone = non_json_rpc_map.clone(); @@ -106,13 +106,18 @@ impl WebsocketBroker { } else { let cleaner_clone = non_json_rpc_map.clone(); tokio::spawn(async move { - while let Some(v) = cleaner_tr.recv().await { + while let Some(cleanup_type) = cleaner_tr.recv().await { + // Extract the ID from the cleanup type + let id = match cleanup_type { + CleanupType::Connection(ref cid) => cid.clone(), + CleanupType::Session(ref sid) => sid.clone(), + }; { if let Some(cleaner_list) = - { cleaner_clone.write().unwrap().remove(&v) } + { cleaner_clone.write().unwrap().remove(&id) } { for sender in cleaner_list { - if sender.try_send(v.clone()).is_err() { + if sender.try_send(id.clone()).is_err() { error!("Cleaning up listener"); } } @@ -122,7 +127,12 @@ impl WebsocketBroker { }); while let Some(v) = tr.recv().await { - let id = v.get_id(); + let id = v + .rpc + .ctx + .cid + .clone() + .unwrap_or_else(|| v.rpc.ctx.session_id.clone()); let cleaner = WSNotificationBroker::start( v.clone(), callback.clone(), @@ -391,7 +401,13 @@ mod tests { broker.sender.send(request).await.unwrap(); - broker.cleaner.cleaner.unwrap().send(id).await.unwrap(); + broker + .cleaner + .cleaner + .unwrap() + .send(CleanupType::Connection(id)) + .await + .unwrap(); // See if ws is closed assert!(tr.recv().await.unwrap()) } diff --git a/core/main/src/firebolt/firebolt_gateway.rs b/core/main/src/firebolt/firebolt_gateway.rs index 4470cd34a..1ee8f73b9 100644 --- a/core/main/src/firebolt/firebolt_gateway.rs +++ b/core/main/src/firebolt/firebolt_gateway.rs @@ -18,7 +18,6 @@ use jsonrpsee::{core::server::rpc_module::Methods, types::TwoPointZero}; use ripple_sdk::{ api::{ - device::device_events::{DeviceEvent, DeviceEventCallback, DeviceEventRequest}, firebolt::{ fb_capabilities::JSON_RPC_STANDARD_ERROR_INVALID_PARAMS, fb_openrpc::FireboltOpenRpcMethod, @@ -33,7 +32,7 @@ use ripple_sdk::{ }, chrono::Utc, extn::extn_client_message::ExtnMessage, - log::{debug, error, info, trace, warn}, + log::{error, info, trace, warn}, serde_json::{self, Value}, service::service_message::{JsonRpcMessage as JsonRpcServiceMessage, ServiceMessage}, tokio::{self, runtime::Handle, sync::mpsc::Sender}, @@ -82,6 +81,7 @@ pub enum FireboltGatewayCommand { session_id: String, session: Session, }, + /// Cleanup when a WebSocket connection disconnects UnregisterSession { session_id: String, cid: String, @@ -141,57 +141,26 @@ impl FireboltGateway { } UnregisterSession { session_id, cid } => { info!( - "Cleanup: app disconnect - removing event listeners, broker subs, session" + "Cleanup: WebSocket disconnect - cid={}, session_id={}", + cid, session_id ); - // Clean event listeners by session_id - AppEvents::remove_session(&self.state.platform_state, session_id.clone()); - // Also clean event listeners by connection_id (cid) in case session_id != cid + // Clean event listeners by connection_id (cid). + // All WebSocket RPC handlers populate cid, so we clean only this connection's listeners. + // removes only this connection's listeners, leaving others intact. AppEvents::cleanup_by_connection_id(&self.state.platform_state, &cid); ProviderBroker::unregister_session(&self.state.platform_state, cid.clone()) .await; + + // Cleanup broker subscriptions by connection_id (cid) + // This removes only subscriptions made by this specific WebSocket connection self.state .platform_state .endpoint_state - .cleanup_for_app(&cid) - .await; - // Also cleanup broker subscriptions by session_id (subscription_map uses session_id as key) - self.state - .platform_state - .endpoint_state - .cleanup_for_app(&session_id) + .cleanup_for_connection(&cid) .await; - // Resolve app_id from session state BEFORE clearing the session, - // because ThunderEventProcessor stores listeners by app_id (e.g. "epg"), - // not by cid (a UUID). - let app_id_for_cleanup = self - .state - .platform_state - .session_state - .get_app_id(cid.clone()); + // Clear session from session_map by connection_id (the key used during registration) self.state.platform_state.session_state.clear_session(&cid); - // Send cleanup to ThunderEventProcessor (extension side) to remove - // event_map and last_event entries for this app - if let Some(app_id) = app_id_for_cleanup { - let cleanup_request = DeviceEventRequest { - event: DeviceEvent::Cleanup, - subscribe: false, - callback_type: DeviceEventCallback::FireboltAppEvent(app_id.clone()), - }; - if let Err(e) = self - .state - .platform_state - .get_client() - .send_extn_request_transient(cleanup_request) - { - warn!( - "Failed to send ThunderEventProcessor cleanup for app_id={}: {:?}", - app_id, e - ); - } - } else { - debug!("Could not resolve app_id, skipping ThunderEventProcessor cleanup"); - } } HandleRpc { request } => self.handle(request, None).await, HandleRpcForExtn { msg } => { diff --git a/core/main/src/service/apps/app_events.rs b/core/main/src/service/apps/app_events.rs index 0158aaa8c..a58cc7f40 100644 --- a/core/main/src/service/apps/app_events.rs +++ b/core/main/src/service/apps/app_events.rs @@ -635,4 +635,71 @@ pub mod tests { assert_eq!(state.app_events_state.listeners.read().unwrap().len(), 1); } + + #[tokio::test] + async fn test_cleanup_by_connection_id_shared_session_multi_connection() { + // CRITICAL TEST: Secure mode scenario where multiple connections share the same session_id. + // This verifies that cleanup_by_connection_id(cid) only removes the specific connection's listeners, + // NOT all listeners for the shared session_id. + // + // Background: In secure mode (ws://127.0.0.1:3473?session=), + // multiple WebSocket connections can share the same session_id but have unique cid values. + // If cleanup incorrectly removes by session_id instead of cid, it would over-clean. + let state = PlatformState::mock(); + let session_abc = "session-abc"; + let app = "app1"; + + // Connection 1 and 2 both join the same secure session "session-abc" + let ctx1 = make_ctx(session_abc, app, Some("conn-1-uuid")); + let ctx2 = make_ctx(session_abc, app, Some("conn-2-uuid")); + + // Both subscribe to the same event + add_listener_direct(&state, "onNameChanged", ctx1, None); + add_listener_direct(&state, "onNameChanged", ctx2, None); + + // Verify both listeners are registered + { + let listeners = state.app_events_state.listeners.read().unwrap(); + assert_eq!( + listeners + .get("onNameChanged") + .unwrap() + .get(&None) + .unwrap() + .len(), + 2 + ); + } + + // Connection 1 disconnects: cleanup only by conn-1-uuid, NOT by session_id + AppEvents::cleanup_by_connection_id(&state, "conn-1-uuid"); + + // Verify ONLY Connection 1's listener was removed, Connection 2's remains + { + let listeners = state.app_events_state.listeners.read().unwrap(); + assert_eq!( + listeners + .get("onNameChanged") + .unwrap() + .get(&None) + .unwrap() + .len(), + 1 + ); + assert_eq!( + listeners.get("onNameChanged").unwrap().get(&None).unwrap()[0] + .call_ctx + .cid + .as_deref(), + Some("conn-2-uuid") + ); + // Verify Connection 2 still has the correct session_id + assert_eq!( + listeners.get("onNameChanged").unwrap().get(&None).unwrap()[0] + .call_ctx + .session_id, + session_abc + ); + } + } } diff --git a/core/main/src/service/apps/delegated_launcher_handler.rs b/core/main/src/service/apps/delegated_launcher_handler.rs index 91d14aa54..952d3c7c3 100644 --- a/core/main/src/service/apps/delegated_launcher_handler.rs +++ b/core/main/src/service/apps/delegated_launcher_handler.rs @@ -1308,10 +1308,19 @@ impl DelegatedLauncherHandler { async fn end_session(&mut self, app_id: &str) -> Result { debug!("end_session: entry: app_id={}", app_id); let app = self.platform_state.app_manager_state.remove(app_id); - if app.is_some() { + if let Some(app) = app { if let Some(timer) = self.timer_map.remove(app_id) { timer.cancel(); } + // Clean up all subscriptions for this session when the app session ends + debug!( + "end_session: cleaning up subscriptions for session_id={}", + app.session_id + ); + self.platform_state + .endpoint_state + .cleanup_for_session(&app.session_id) + .await; } else { error!("end_session app_id={} Not found", app_id); return Err(AppError::NotFound); diff --git a/core/sdk/src/api/device/device_events.rs b/core/sdk/src/api/device/device_events.rs index 52eb57e02..29858860e 100644 --- a/core/sdk/src/api/device/device_events.rs +++ b/core/sdk/src/api/device/device_events.rs @@ -39,9 +39,6 @@ pub const VOICE_GUIDANCE_SPEED_CHANGED: &str = "voiceguidance.onSpeedChanged"; pub enum DeviceEvent { InputChanged, AudioChanged, - /// Signals that an app has disconnected and its event subscriptions should be cleaned up. - /// The callback_type.get_id() carries the app_id to clean. - Cleanup, } impl FromStr for DeviceEvent { @@ -51,7 +48,6 @@ impl FromStr for DeviceEvent { match s { "device.onHdcpChanged" => Ok(Self::InputChanged), "device.onAudioChanged" => Ok(Self::AudioChanged), - "device.cleanup" => Ok(Self::Cleanup), _ => Err(()), } } @@ -95,7 +91,6 @@ impl ExtnPayloadProvider for DeviceEventRequest { match self.event { DeviceEvent::InputChanged => RippleContract::DeviceEvents(EventAdjective::Input), DeviceEvent::AudioChanged => RippleContract::DeviceEvents(EventAdjective::Audio), - DeviceEvent::Cleanup => RippleContract::DeviceEvents(EventAdjective::Input), } } @@ -112,7 +107,6 @@ mod tests { #[rstest(input, expected, case("device.onHdcpChanged", Ok(DeviceEvent::InputChanged)), - case("device.cleanup", Ok(DeviceEvent::Cleanup)), case("invalid_event", Err(())), )] fn test_from_str(input: &str, expected: Result) { @@ -120,50 +114,13 @@ mod tests { } #[test] - fn test_cleanup_event_request_payload_roundtrip() { + fn test_device_event_request_payload_roundtrip() { let request = DeviceEventRequest { - event: DeviceEvent::Cleanup, - subscribe: false, - callback_type: DeviceEventCallback::FireboltAppEvent("epg".to_string()), - }; - let payload = request.get_extn_payload(); - let recovered = DeviceEventRequest::get_from_payload(payload).unwrap(); - assert_eq!(recovered.event, DeviceEvent::Cleanup); - assert_eq!(recovered.callback_type.get_id(), "epg"); - } - - #[test] - fn test_cleanup_event_contract() { - let request = DeviceEventRequest { - event: DeviceEvent::Cleanup, - subscribe: true, - callback_type: DeviceEventCallback::FireboltAppEvent("app1".to_string()), - }; - // Cleanup routes through the same contract as Input events - assert_eq!( - request.get_contract(), - RippleContract::DeviceEvents(EventAdjective::Input) - ); - } - - #[rstest] - #[case( - DeviceEventCallback::FireboltAppEvent("app_event".to_string()), - "app_event" - )] - #[case(DeviceEventCallback::ExtnEvent, "internal")] - fn test_get_id(#[case] callback: DeviceEventCallback, #[case] expected_id: &str) { - assert_eq!(callback.get_id(), expected_id); - } - - #[test] - fn test_extn_request_device_event() { - let device_event_request = DeviceEventRequest { event: DeviceEvent::InputChanged, subscribe: true, callback_type: DeviceEventCallback::FireboltAppEvent("id".to_string()), }; let contract_type: RippleContract = RippleContract::DeviceEvents(EventAdjective::Input); - test_extn_payload_provider(device_event_request, contract_type); + test_extn_payload_provider(request, contract_type); } } diff --git a/core/sdk/src/service/service_client.rs b/core/sdk/src/service/service_client.rs index 5416c94e4..eb651b505 100644 --- a/core/sdk/src/service/service_client.rs +++ b/core/sdk/src/service/service_client.rs @@ -245,7 +245,7 @@ impl ServiceClient { let handle_ws_message = |msg: Message| { if let Message::Text(message) = msg.clone() { // Service message - debug!("Received Service Message: {:#?}", message); + trace!("Received Service Message: {:#?}", message); if let Ok(sm) = serde_json::from_str::(&message) { match sm.message { JsonRpcMessage::Request(ref _json_rpc_request) => { @@ -263,7 +263,7 @@ impl ServiceClient { } } JsonRpcMessage::Notification(ref json_rpc_notification) => { - debug!( + trace!( "Received Service Notification: {:?}", json_rpc_notification, ); diff --git a/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs b/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs index 9c68c75a2..52efecff9 100644 --- a/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs +++ b/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs @@ -93,10 +93,6 @@ impl ThunderEventMessage { value.clone(), ))) } - DeviceEvent::Cleanup => { - // Cleanup is not a real Thunder event, skip - return None; - } } } else { debug!( diff --git a/device/thunder_ripple_sdk/src/processors/thunder_events.rs b/device/thunder_ripple_sdk/src/processors/thunder_events.rs index f0128bc6b..43a62329b 100644 --- a/device/thunder_ripple_sdk/src/processors/thunder_events.rs +++ b/device/thunder_ripple_sdk/src/processors/thunder_events.rs @@ -16,7 +16,7 @@ // use ripple_sdk::{ - api::session::EventAdjective, framework::ripple_contract::RippleContract, log::info, + api::session::EventAdjective, framework::ripple_contract::RippleContract, utils::error::RippleError, }; @@ -106,19 +106,6 @@ impl ExtnRequestProcessor for ThunderOpenEventsProcessor { id.clone(), HDCPEventHandler::provide(id, callback_type), )), - DeviceEvent::Cleanup => { - // Clean up all event subscriptions for this app - let removed_events = state.event_processor.cleanup_by_app_id(&id); - if !removed_events.is_empty() { - info!( - "ThunderEventProcessor cleanup: removed {} event(s) for app {}: {:?}", - removed_events.len(), - id, - removed_events - ); - } - return true; - } } { v.await; Self::ack(state.get_client(), msg).await.is_ok()